(spark) branch master updated: [SPARK-56842][SQL] Short-circuit AQE when materialized stages are empty

[sunchao]

This is an automated email from the ASF dual-hosted git repository.

sunchao 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 688a73cd32a4 [SPARK-56842][SQL] Short-circuit AQE when materialized 
stages are empty
688a73cd32a4 is described below

commit 688a73cd32a4d3fdf8446053ded2049fa47b5487
Author: Chao Sun <c...@openai.com>
AuthorDate: Fri Aug 21 10:06:08 2026 -0700

    [SPARK-56842][SQL] Short-circuit AQE when materialized stages are empty
    
    ### Why are the changes needed?
    
    Adaptive Query Execution can miss a provable empty or nonempty result when 
a materialized query stage is represented through a supported physical sort, 
limit, or aggregate operator. In particular, a global aggregate over empty 
input still produces one output row: treating that aggregate as empty changes 
the result of conditionless semi/anti joins.
    
    Adaptive replanning can also remove exchange stages that are still 
materializing. Leaving a private obsolete stage running wastes work and can let 
an irrelevant materialization failure abort the selected query plan. Reusing 
the same exchange inside one adaptive plan must not permanently protect that 
result after every local alias disappears, while reuse by another adaptive plan 
or subquery must remain protected. Some obsolete stages cannot safely be 
cancelled at all, and an opaque del [...]
    
    ### What changes were proposed in this PR?
    
    - Infer runtime row counts only through the supported, explicitly safe 
`SortExec`, `BaseLimitExec`, and `BaseAggregateExec` cases. A global aggregate 
over an empty stage reports one row; grouped aggregates preserve the child 
estimate; limits propagate only provable emptiness, including `LIMIT 0`.
    - Remove the unsupported `AQEShuffleReadExec`, `ProjectExec`, and 
`ColumnarToRowExec` recursion paths instead of claiming coverage for physical 
shapes that the adaptive rule cannot reach.
    - Derive obsolete-stage candidates from both the previous physical plan and 
the current round of replacement stages, deduplicating stage IDs before 
applying the unfinished/private-exchange checks so an older stage remains 
discoverable when a later replan drops its branch.
    - Record the owning adaptive-plan instance when an exchange result enters 
the shared stage cache. Reuse within that same plan remains cancellable once 
every local alias disappears; reuse by a different plan or subquery, or reuse 
of an entry without a recorded owner, remains permanently protected. Compare 
owners by reference identity rather than the plan's structural `equals` 
implementation.
    - Track every historical local stage ID associated with a reused result, 
group cancellation by result identity, and cancel its underlying exchange at 
most once. Return or mark all alias IDs so every materialization callback 
suppresses an irrelevant ordinary failure while retained/shared and fatal 
failures continue to propagate.
    - Permanently remember a failed cancellation by result identity and never 
retry it for a later same-plan reuse alias: a native shuffle marks itself 
cancelled before attempting to cancel its `FutureAction`, so a retry could 
otherwise report false success while its job is still running. Keep the 
original cache entry available and track later aliases for the same guarded 
obsolete-failure handling.
    - Proactively cancel only eligible, private obsolete shuffle stages. Never 
cancel broadcast stages, stages already shared or retained by the adopted plan, 
or opaque delegated shuffle exchanges that cannot expose their own cancellable 
submission. Mark only private obsolete broadcasts and identifiable submitted 
shuffles that cannot safely be cancelled for guarded eventual-failure handling; 
opaque delegated shuffles are neither cancelled nor marked for failure 
suppression because their h [...]
    - Register an already-submitted shuffle ID before any cancellation attempt, 
without creating a lazy shuffle dependency for an unsubmitted stage. Resolve 
the underlying exchange directly from `ShuffleQueryStageExec.shuffle`, 
including aliases wrapped in the leaf `ReusedExchangeExec`. Under 
`RemoveShuffleFiles`, do not cancel a submitted shuffle: scheduler cancellation 
is asynchronous, so deleting its files could race map writers. Submitted 
shuffles remain eligible under cleanup modes t [...]
    - Preserve the exchange-cache entry when cancellation fails or an obsolete 
stage cannot safely be cancelled. Remove it immediately only after successful 
cancellation. For a failed cancellation, an obsolete broadcast, or a known 
submitted shuffle protected by `RemoveShuffleFiles`, ignore an eventual 
nonfatal failure only after confirming that the stage remains private and 
absent from the current physical plan, registering any visible submitted 
shuffle ID, and evicting the failed cache  [...]
    - Serialize exchange-cache lookup, owner registration, reuse, and 
cross-plan shared-result marking under short adaptive stage-lifecycle lock 
sections. Reserve cancellation using a stage-result-scoped `CompletableFuture`, 
then acquire the shuffle monitor and cancel outside the query-global lock. Both 
the initial cache lookup and the post-child-planning cache lookup await 
cancellation of that same result outside the global lock and retry; unrelated 
exchanges continue. Keep stage-local s [...]
    - Precompute retained stage IDs and shared-result identities and remove 
exchange-cache entries through canonical-key lookup, avoiding repeated 
full-cache scans during adaptive replanning.
    - Document nested-stage materialization, same-plan versus cross-subquery 
exchange ownership, lifecycle lock ordering, alias failure callbacks, and the 
permanent protection of externally shared stage results. Keep lifecycle helpers 
scoped to `private[adaptive]`.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes. Eligible obsolete shuffle jobs may be cancelled under cleanup modes 
that do not remove submitted shuffle files, and listeners may observe 
corresponding failed/cancelled job events. Broadcast jobs, opaque delegated 
shuffles, and submitted shuffles subject to immediate file removal are 
intentionally left running. An eventual ordinary failure from a private 
obsolete broadcast or identifiable cleanup-protected shuffle may be ignored 
after cleanup checks; an opaque delegated shuffle's [...]
    
    ### How was this PR tested?
    
    The latest owner-aware exchange reuse, alias-wide failure handling, direct 
shuffle-registration, and sticky failed-cancellation changes passed both 
validation runs on top of published PR head 
`4baed923d832a9452d48d6a08a97be04c06e73e6`. The owner-aware follow-up remains 
local, uncommitted, and unpublished.
    
    - Focused exchange-reuse regressions: **13 tests passed**, with **0 
failures, 0 errors, and 0 skipped tests**; JUnit XML timestamp 
`2026-08-19T14:03:59`, duration `4.893s`.
    - Complete `AdaptiveQueryExecSuite`: **144 tests passed**, with **0 
failures, 0 errors, and 0 skipped tests**; JUnit XML timestamp 
`2026-08-19T14:07:18`, duration `127.294s`.
    
    The regression coverage includes:
    
    - A mutation-sensitive conditionless left-semi join over a global aggregate 
on empty input, with assertions for the original join and its adaptive rewrite.
    - A real planned `ORDER BY ... LIMIT ... DISTINCT` query that establishes a 
logically linked physical sort/shuffle shape, plus direct rule assertions for 
sort and limit wrappers derived from that shape.
    - Zero and positive limits, offsets, unknown statistics, grouped 
aggregates, and the single output row of global aggregates over empty input.
    - Exchange reuse, private/shared-stage protection, cache retention after 
failed cancellation, fatal-stage failure propagation, and broadcast 
cancellation safety.
    - Same-plan reuse with multiple dropped aliases, one underlying 
cancellation, and an ordinary failure callback for each alias; retained aliases 
and fatal alias failures remain protected.
    - Reuse by a separately owned adaptive subquery remains protected even when 
its plan compares structurally equal to the original owner; cache entries with 
unknown owners also remain protected.
    - A failed cancellation followed by later same-plan reuse keeps the 
original exchange cached, never retries the already-marked native cancellation, 
and guards eventual failures from the additional aliases.
    - Historical same-plan alias IDs remain tracked when an earlier 
adaptive-plan adoption has already dropped those aliases; subsequent 
cancellation returns every alias ID and suppresses each obsolete ordinary 
failure.
    - Submitted-shuffle ID registration, including a reused-exchange alias 
backed by its actual owner-created shuffle, unsubmitted lazy-dependency safety, 
and `RemoveShuffleFiles` protection against active map writers. A 
submitted-shuffle regression checks that a wrapped fatal failure still 
propagates, a later ordinary obsolete-stage failure is ignored, its shuffle ID 
is registered and its failed cache entry is removed, and its underlying shuffle 
action is never cancelled.
    - An opaque delegated exchange whose own `futureAction` is empty remains 
uncancelled and cached; its later ordinary failure is not suppressed because 
the hidden delegate's shuffle ID cannot safely be registered.
    - A stage present in a constructed previous physical plan but absent from 
the latest replacement batch, including deduplication, cancellation, later 
nonfatal cancelled-stage failure suppression, and cache retention for the stage 
kept by the new plan.
    - Unrelated-exchange progress while either a generic cancellation callback 
or a real shuffle monitor is blocked; same-result reuse waits outside the 
global lock and reuses the original cached stage if cancellation fails.
    
    The focused regressions and complete suite were run against the updated 
integration worktree:
    
    ```bash
    /private/tmp/spark-pr55839-sbt.sh \
      'sql/testOnly 
org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite -- -z "reuse"'
    /private/tmp/spark-pr55839-sbt.sh \
      'sql/testOnly 
org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite'
    ```
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: OpenAI Codex
    
    Closes #55839 from sunchao/dev/chao/codex/oss-aqe-empty-stage.
    
    Lead-authored-by: Chao Sun <c...@openai.com>
    Co-authored-by: fredliu-data <fred...@openai.com>
    Signed-off-by: Chao Sun <c...@openai.com>
---
 .../adaptive/AQEPropagateEmptyRelation.scala       |   37 +-
 .../execution/adaptive/AdaptiveSparkPlanExec.scala |  355 ++++++-
 .../adaptive/AdaptiveQueryExecSuite.scala          | 1099 +++++++++++++++++++-
 3 files changed, 1469 insertions(+), 22 deletions(-)

diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala
index e2a013b9e814..d9107c6a128e 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEPropagateEmptyRelation.scala
@@ -22,6 +22,7 @@ import 
org.apache.spark.sql.catalyst.planning.ExtractSingleColumnNullAwareAntiJo
 import org.apache.spark.sql.catalyst.plans.logical.EmptyRelation
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.trees.TreePattern.{LOCAL_RELATION, 
LOGICAL_QUERY_STAGE, TRUE_OR_FALSE_LITERAL}
+import org.apache.spark.sql.execution.{BaseLimitExec, SortExec, SparkPlan}
 import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
 import org.apache.spark.sql.execution.exchange.{REPARTITION_BY_COL, 
REPARTITION_BY_NUM, ShuffleExchangeLike}
 import org.apache.spark.sql.execution.joins.HashedRelationWithAllNullKeys
@@ -52,19 +53,39 @@ object AQEPropagateEmptyRelation extends 
PropagateEmptyRelationBase {
   //   - positive value means an estimated row count which can be 
over-estimated
   //   - none means the plan has not materialized or the plan can not be 
estimated
   private def getEstimatedRowCount(plan: LogicalPlan): Option[BigInt] = plan 
match {
-    case LogicalQueryStage(_, stage: QueryStageExec) if stage.isMaterialized =>
+    case LogicalQueryStage(_, physicalPlan) =>
+      getEstimatedRowCount(physicalPlan)
+
+    case _: EmptyRelation => Some(0)
+
+    case _ => None
+  }
+
+  private def getEstimatedRowCount(plan: SparkPlan): Option[BigInt] = plan 
match {
+    case stage: QueryStageExec if stage.isMaterialized =>
       stage.getRuntimeStatistics.rowCount
 
-    case LogicalQueryStage(_, agg: BaseAggregateExec) if 
agg.groupingExpressions.nonEmpty &&
-      agg.child.isInstanceOf[QueryStageExec] =>
-      val stage = agg.child.asInstanceOf[QueryStageExec]
-      if (stage.isMaterialized) {
-        stage.getRuntimeStatistics.rowCount
+    case sort: SortExec =>
+      getEstimatedRowCount(sort.child)
+
+    // A global limit can also drop rows through an offset, so only propagate
+    // proven emptiness; a zero limit is unconditionally empty.
+    case limit: BaseLimitExec =>
+      if (limit.limit == 0) {
+        Some(BigInt(0))
       } else {
-        None
+        getEstimatedRowCount(limit.child).filter(_ == 0)
       }
 
-    case _: EmptyRelation => Some(0)
+    // Match the global-aggregate invariant in LogicalQueryStage.computeStats 
without adopting
+    // its logical-statistics fallback or its unrestricted physical-plan 
traversal.
+    case aggregate: BaseAggregateExec if aggregate.groupingExpressions.isEmpty 
=>
+      getEstimatedRowCount(aggregate.child).map { rowCount =>
+        if (rowCount == 0) BigInt(1) else rowCount
+      }
+
+    case aggregate: BaseAggregateExec =>
+      getEstimatedRowCount(aggregate.child)
 
     case _ => None
   }
diff --git 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
index bfe6a9a3f633..a12b202b5b00 100644
--- 
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
+++ 
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
@@ -18,7 +18,8 @@
 package org.apache.spark.sql.execution.adaptive
 
 import java.util
-import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue}
+import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, 
LinkedBlockingQueue}
+import java.util.concurrent.atomic.AtomicReference
 
 import scala.collection.concurrent.TrieMap
 import scala.collection.mutable
@@ -77,6 +78,17 @@ case class AdaptiveSparkPlanExec(
 
   @transient private val lock = new Object()
 
+  // Access is serialized by the execution context's stage lifecycle lock.
+  @transient private val uncancelledObsoleteStageIds = 
mutable.HashSet.empty[Int]
+
+  // Remember every local alias so cancellation and failure handling cover 
every stage ID.
+  @transient private val stageIdsByResult =
+    mutable.HashMap.empty[AtomicReference[Option[Any]], 
mutable.LinkedHashSet[Int]]
+
+  // A failed native cancellation can leave its exchange marked cancelled 
while its job still runs.
+  @transient private val failedStageCancellationResults =
+    mutable.HashSet.empty[AtomicReference[Option[Any]]]
+
   @transient private val logOnLevel: ( => MessageWithContext) => Unit =
     logBasedOnLevel(conf.adaptiveExecutionLogLevel)
 
@@ -301,6 +313,7 @@ case class AdaptiveSparkPlanExec(
         var result = createQueryStages(fun, currentPhysicalPlan, firstRun = 
true)
         val events = new LinkedBlockingQueue[StageMaterializationEvent]()
         val errors = new mutable.ArrayBuffer[Throwable]()
+        val obsoleteCancelledStageIds = new mutable.HashSet[Int]
         var stagesToReplace = Seq.empty[QueryStageExec]
         while (!result.allChildStagesMaterialized) {
           currentPhysicalPlan = result.newPlan
@@ -356,7 +369,9 @@ case class AdaptiveSparkPlanExec(
               stage.resultOption.set(Some(res))
             case StageFailure(stage, ex) =>
               stage.error.set(Some(ex))
-              errors.append(ex)
+              if (!shouldIgnoreObsoleteStageFailure(stage, ex, 
obsoleteCancelledStageIds)) {
+                errors.append(ex)
+              }
           }
 
           // In case of errors, we cancel all running stages and throw 
exception.
@@ -397,6 +412,9 @@ case class AdaptiveSparkPlanExec(
                   currentPhysicalPlan.treeString, 
newPhysicalPlan.treeString).mkString("\n")
                 logOnLevel(log"Plan changed:\n${MDC(QUERY_PLAN, plans)}")
                 cleanUpTempTags(newPhysicalPlan)
+                obsoleteCancelledStageIds ++=
+                  cancelObsoleteStages(
+                    newPhysicalPlan, 
obsoleteStageCandidates(currentPhysicalPlan, stagesToReplace))
                 currentPhysicalPlan = newPhysicalPlan
                 currentLogicalPlan = newLogicalPlan
                 stagesToReplace = Seq.empty[QueryStageExec]
@@ -420,6 +438,204 @@ case class AdaptiveSparkPlanExec(
       .get.asInstanceOf[T]
   }
 
+  /** Include stages retained by earlier adopted plans as well as newly 
created query stages. */
+  private def obsoleteStageCandidates(
+      oldPhysicalPlan: SparkPlan,
+      stagesToReplace: Seq[QueryStageExec]): Seq[QueryStageExec] = {
+    val visitedStageIds = mutable.HashSet.empty[Int]
+    (oldPhysicalPlan.collect { case stage: QueryStageExec => stage } ++ 
stagesToReplace)
+      .filter(stage => visitedStageIds.add(stage.id))
+  }
+
+  /**
+   * Cancel unfinished exchange stages that are no longer referenced by the 
adopted plan.
+   *
+   * An unfinished stage cannot be hidden inside another stage's plan because 
parent stages are
+   * created only after all child stages have materialized. Results reused by 
another adaptive
+   * plan, such as a subquery, are separately protected by the context's 
shared-result tracking.
+   * A failed or unsafe cancellation leaves the stage cached until its 
eventual materialization
+   * event proves whether the stage is still required.
+   */
+  private def cancelObsoleteStages(
+      newPhysicalPlan: SparkPlan,
+      candidateStages: Seq[QueryStageExec]): Seq[Int] = {
+    val retainedStageIds = mutable.HashSet.empty[Int]
+    val retainedStageResults = 
mutable.HashSet.empty[AtomicReference[Option[Any]]]
+    newPhysicalPlan.foreach {
+      case stage: QueryStageExec =>
+        retainedStageIds += stage.id
+        retainedStageResults += stage.resultOption
+      case _ =>
+    }
+    val obsoleteStages = mutable.LinkedHashMap.empty[
+      AtomicReference[Option[Any]], 
mutable.ArrayBuffer[ExchangeQueryStageExec]]
+    candidateStages.foreach {
+      case stage: ExchangeQueryStageExec
+          if !retainedStageIds.contains(stage.id) &&
+            !retainedStageResults.contains(stage.resultOption) =>
+        obsoleteStages.getOrElseUpdate(stage.resultOption, 
mutable.ArrayBuffer.empty) += stage
+      case _ =>
+    }
+    obsoleteStages.values.toSeq.flatMap { stages =>
+      val stage = stages.head
+      val reservation = context.withStageLifecycleLock {
+        if (!stage.isMaterialized && 
!context.isSharedStageResult(stage.resultOption)) {
+          val recordedStageIds = 
stageIdsByResult.get(stage.resultOption).toSeq.flatten
+          val obsoleteStageIds = (recordedStageIds ++ 
stages.map(_.id)).distinct
+          if (failedStageCancellationResults.contains(stage.resultOption)) {
+            uncancelledObsoleteStageIds ++= obsoleteStageIds
+            None
+          } else {
+            context.reserveStageCancellation(stage.resultOption).map { 
cancellation =>
+              (cancellation, obsoleteStageIds)
+            }
+          }
+        } else {
+          None
+        }
+      }
+      reservation.toSeq.flatMap { case (cancellation, obsoleteStageIds) =>
+        try {
+          withShuffleCancellationLock(stage) {
+            val shouldCancel = context.withStageLifecycleLock {
+              if (stage.isMaterialized || 
context.isSharedStageResult(stage.resultOption)) {
+                false
+              } else if (canCancelStage(stage)) {
+                true
+              } else {
+                if (canTrackObsoleteStageFailure(stage)) {
+                  uncancelledObsoleteStageIds ++= obsoleteStageIds
+                }
+                false
+              }
+            }
+            if (shouldCancel) {
+              try {
+                recordSubmittedShuffleIds(stage)
+                stage.cancel(
+                  "The query stage is no longer referenced by the current 
adaptive plan.")
+                context.withStageLifecycleLock {
+                  removeStageFromCache(stage)
+                }
+                obsoleteStageIds
+              } catch {
+                case NonFatal(t) =>
+                  context.withStageLifecycleLock {
+                    failedStageCancellationResults += stage.resultOption
+                    uncancelledObsoleteStageIds ++= obsoleteStageIds
+                  }
+                  logError(s"Exception in cancelling obsolete query stage: 
${stage.treeString}", t)
+                  Seq.empty
+              }
+            } else {
+              Seq.empty
+            }
+          }
+        } finally {
+          context.withStageLifecycleLock {
+            context.finishStageCancellation(stage.resultOption, cancellation)
+          }
+        }
+      }
+    }
+  }
+
+  /** Serialize shuffle eligibility and cancellation with the exchange's job 
submission. */
+  private def withShuffleCancellationLock[T](
+      stage: ExchangeQueryStageExec)(body: => T): T = stage match {
+    case shuffleStage: ShuffleQueryStageExec => 
shuffleStage.shuffle.synchronized(body)
+    case _ => body
+  }
+
+  /**
+   * Broadcast cancellation cannot prevent its supplier from submitting 
another job afterward.
+   * Delegating shuffles may conceal an already submitted job behind an empty 
futureAction, and
+   * submitted shuffle cancellation is asynchronous, so removing its files can 
race map writers.
+   */
+  private def canCancelStage(stage: ExchangeQueryStageExec): Boolean = stage 
match {
+    case _: BroadcastQueryStageExec => false
+    case shuffleStage: ShuffleQueryStageExec =>
+      val shuffle = shuffleStage.shuffle
+      val submitted = shuffle.futureAction.get().isDefined
+      (shuffle.isInstanceOf[ShuffleExchangeExec] || submitted) &&
+        (!submitted || context.qe.shuffleCleanupMode != RemoveShuffleFiles)
+    case _ => true
+  }
+
+  /** Opaque delegated shuffles can conceal submitted jobs and their required 
cleanup IDs. */
+  private def canTrackObsoleteStageFailure(stage: ExchangeQueryStageExec): 
Boolean = stage match {
+    case shuffleStage: ShuffleQueryStageExec =>
+      shuffleStage.shuffle.isInstanceOf[ShuffleExchangeExec] ||
+        shuffleStage.shuffle.futureAction.get().isDefined
+    case _ => true
+  }
+
+  /** Never hide fatal stage failures, even when their stage was cancelled as 
obsolete. */
+  private def shouldIgnoreObsoleteStageFailure(
+      stage: QueryStageExec,
+      error: Throwable,
+      obsoleteCancelledStageIds: scala.collection.Set[Int]): Boolean = {
+    !error.isInstanceOf[SparkFatalException] && NonFatal(error) &&
+      (obsoleteCancelledStageIds.contains(stage.id) || 
ignoreFailedObsoleteStageFailure(stage))
+  }
+
+  /**
+   * Suppress an uncancelled obsolete stage's later failure only while its 
result remains private
+   * and absent from the current physical plan. This covers failed 
cancellation and stages whose
+   * cancellation would be unsafe. Reused stages must still fail their 
consumers; otherwise, the
+   * failed cache entry can now be safely discarded.
+   */
+  private def ignoreFailedObsoleteStageFailure(stage: QueryStageExec): Boolean 
= stage match {
+    case exchangeStage: ExchangeQueryStageExec =>
+      context.withStageLifecycleLock {
+        if (!uncancelledObsoleteStageIds.remove(stage.id)) {
+          false
+        } else {
+          val stillReferenced = currentPhysicalPlan.exists {
+            case retainedStage: QueryStageExec =>
+              retainedStage.id == stage.id || 
retainedStage.resultOption.eq(stage.resultOption)
+            case _ => false
+          }
+          if (!stillReferenced && 
!context.isSharedStageResult(stage.resultOption)) {
+            recordSubmittedShuffleIds(exchangeStage)
+            removeStageFromCache(exchangeStage)
+            true
+          } else {
+            false
+          }
+        }
+      }
+    case _ => false
+  }
+
+  /** Record only submitted shuffles, avoiding initialization of an 
unsubmitted lazy dependency. */
+  private def recordSubmittedShuffleIds(stage: ExchangeQueryStageExec): Unit = 
{
+    def record(shuffle: ShuffleExchangeLike): Unit = {
+      if (shuffle.futureAction.get().isDefined) {
+        context.shuffleIds.put(shuffle.shuffleId, true)
+      }
+    }
+    stage match {
+      // A reuse instance wraps its exchange in ReusedExchangeExec, which is a 
leaf node.
+      case shuffleStage: ShuffleQueryStageExec => record(shuffleStage.shuffle)
+      case _ =>
+        stage.plan.foreach {
+          case shuffle: ShuffleExchangeLike => record(shuffle)
+          case _ =>
+        }
+    }
+  }
+
+  /** Remove the exact canonical exchange-cache entry only when it still owns 
this result. */
+  private def removeStageFromCache(stage: ExchangeQueryStageExec): Unit = {
+    val cacheKey = stage.plan.canonicalized
+    context.stageCache.get(cacheKey).foreach { cachedStage =>
+      if (cachedStage.resultOption.eq(stage.resultOption)) {
+        context.stageCache.remove(cacheKey)
+      }
+    }
+  }
+
   // Use a lazy val to avoid this being called more than once.
   @transient private lazy val finalPlanUpdate: Unit = {
     // Do final plan update after result stage has materialized.
@@ -623,16 +839,25 @@ case class AdaptiveSparkPlanExec(
   private def createNonResultQueryStages(plan: SparkPlan): CreateStageResult = 
plan match {
     case e: Exchange =>
       // First have a quick check in the `stageCache` without having to 
traverse down the node.
-      context.stageCache.get(e.canonicalized) match {
-        case Some(existingStage) if conf.exchangeReuseEnabled =>
-          val stage = reuseQueryStage(existingStage, e)
+      val reusedStage = if (conf.exchangeReuseEnabled) {
+        withStageCacheEntry(e.canonicalized) {
+          _.map { existingStage =>
+            reuseQueryStage(existingStage, e)
+          }
+        }
+      } else {
+        None
+      }
+
+      reusedStage match {
+        case Some(stage) =>
           val isMaterialized = stage.isMaterialized
           CreateStageResult(
             newPlan = stage,
             allChildStagesMaterialized = isMaterialized,
             newStages = if (isMaterialized) Seq.empty else Seq(stage))
 
-        case _ =>
+        case None =>
           val result = createNonResultQueryStages(e.child)
           val newPlan = 
e.withNewChildren(Seq(result.newPlan)).asInstanceOf[Exchange]
           // Create a query stage only when all the child query stages are 
ready.
@@ -642,10 +867,14 @@ case class AdaptiveSparkPlanExec(
               // Check the `stageCache` again for reuse. If a match is found, 
ditch the new stage
               // and reuse the existing stage found in the `stageCache`, 
otherwise update the
               // `stageCache` with the new stage.
-              val queryStage = context.stageCache.getOrElseUpdate(
-                newStage.plan.canonicalized, newStage)
-              if (queryStage.ne(newStage)) {
-                newStage = reuseQueryStage(queryStage, e)
+              val cacheKey = newStage.plan.canonicalized
+              withStageCacheEntry(cacheKey) {
+                case Some(queryStage) =>
+                  newStage = reuseQueryStage(queryStage, e)
+                case None =>
+                  context.registerStageOwner(newStage.resultOption, this)
+                  recordStageId(newStage)
+                  context.stageCache.put(cacheKey, newStage)
               }
             }
             val isMaterialized = newStage.isMaterialized
@@ -685,6 +914,26 @@ case class AdaptiveSparkPlanExec(
       }
   }
 
+  /** Wait for cancellation of this result without blocking unrelated 
exchange-cache access. */
+  @scala.annotation.tailrec
+  private def withStageCacheEntry[T](
+      cacheKey: SparkPlan)(
+      useStage: Option[ExchangeQueryStageExec] => T): T = {
+    val result: Either[CompletableFuture[Unit], T] = 
context.withStageLifecycleLock {
+      val cachedStage = context.stageCache.get(cacheKey)
+      cachedStage.flatMap(stage => 
context.pendingStageCancellation(stage.resultOption)) match {
+        case Some(cancellation) => Left(cancellation)
+        case None => Right(useStage(cachedStage))
+      }
+    }
+    result match {
+      case Left(cancellation) =>
+        cancellation.join()
+        withStageCacheEntry(cacheKey)(useStage)
+      case Right(value) => value
+    }
+  }
+
   private def newResultQueryStage(
       resultHandler: SparkPlan => Any,
       plan: SparkPlan): ResultQueryStageExec = {
@@ -740,12 +989,19 @@ case class AdaptiveSparkPlanExec(
   private def reuseQueryStage(
       existing: ExchangeQueryStageExec,
       exchange: Exchange): ExchangeQueryStageExec = {
+    context.markSharedStageResult(existing.resultOption, this)
     val queryStage = existing.newReuseInstance(currentStageId, exchange.output)
     currentStageId += 1
     setLogicalLinkForNewQueryStage(queryStage, exchange)
+    recordStageId(queryStage)
     queryStage
   }
 
+  /** Track local reuse aliases whose materialization may report the same 
eventual failure. */
+  private def recordStageId(stage: ExchangeQueryStageExec): Unit = {
+    stageIdsByResult.getOrElseUpdate(stage.resultOption, 
mutable.LinkedHashSet.empty) += stage.id
+  }
+
   /**
    * Set the logical node link of the `stage` as the corresponding logical 
node of the `plan` it
    * encloses. If an `plan` has been transformed from a `Repartition`, it 
should have `logicalLink`
@@ -973,7 +1229,6 @@ object AdaptiveSparkPlanExec {
  * The execution context shared between the main query and all sub-queries.
  */
 case class AdaptiveExecutionContext(session: SparkSession, qe: QueryExecution) 
{
-
   /**
    * The subquery-reuse map shared across the entire query.
    */
@@ -986,6 +1241,84 @@ case class AdaptiveExecutionContext(session: 
SparkSession, qe: QueryExecution) {
   val stageCache: TrieMap[SparkPlan, ExchangeQueryStageExec] =
     new TrieMap[SparkPlan, ExchangeQueryStageExec]()
 
+  private val stageLifecycleLock = new Object
+
+  /**
+   * Serialize exchange-cache lookup, reuse, and cancellation reservations 
across subqueries.
+   * Blocking cancellation and waits for a particular result happen outside 
this query-wide lock.
+   */
+  private[adaptive] def withStageLifecycleLock[T](body: => T): T = {
+    stageLifecycleLock.synchronized(body)
+  }
+
+  /** Stage-scoped reservations keep a cancelling result cached without 
blocking other stages. */
+  private val stageCancellationReservations =
+    new ConcurrentHashMap[AtomicReference[Option[Any]], 
CompletableFuture[Unit]]()
+
+  /** Claim a result before waiting for its shuffle monitor or cancelling its 
exchange. */
+  private[adaptive] def reserveStageCancellation(
+      resultOption: AtomicReference[Option[Any]]): 
Option[CompletableFuture[Unit]] = {
+    val cancellation = new CompletableFuture[Unit]()
+    if (stageCancellationReservations.putIfAbsent(resultOption, cancellation) 
== null) {
+      Some(cancellation)
+    } else {
+      None
+    }
+  }
+
+  /** Return the cancellation that a potential consumer of this particular 
result must await. */
+  private[adaptive] def pendingStageCancellation(
+      resultOption: AtomicReference[Option[Any]]): 
Option[CompletableFuture[Unit]] = {
+    Option(stageCancellationReservations.get(resultOption))
+  }
+
+  /** Publish the final cache state before allowing consumers of this result 
to retry lookup. */
+  private[adaptive] def finishStageCancellation(
+      resultOption: AtomicReference[Option[Any]],
+      cancellation: CompletableFuture[Unit]): Unit = {
+    stageCancellationReservations.remove(resultOption, cancellation)
+    cancellation.complete(())
+  }
+
+  /**
+   * The adaptive plan that first cached each result. Identity, rather than 
case-class equality,
+   * distinguishes independently planned subqueries with equivalent physical 
input plans.
+   */
+  private val stageResultOwners =
+    new ConcurrentHashMap[AtomicReference[Option[Any]], 
AdaptiveSparkPlanExec]()
+
+  /** Record the original owner atomically with exchange-cache insertion. */
+  private[adaptive] def registerStageOwner(
+      resultOption: AtomicReference[Option[Any]], owner: 
AdaptiveSparkPlanExec): Unit = {
+    stageResultOwners.putIfAbsent(resultOption, owner)
+  }
+
+  /**
+   * Results reused by another adaptive plan in this execution context. 
Cross-plan protection
+   * remains for the context's lifetime; aliases within one plan are not 
considered shared.
+   */
+  private val sharedStageResults =
+    new ConcurrentHashMap[AtomicReference[Option[Any]], Boolean]()
+
+  /** Conservatively protect a result whose owner is unknown or whose 
protection is test-forced. */
+  private[adaptive] def markSharedStageResult(resultOption: 
AtomicReference[Option[Any]]): Unit = {
+    sharedStageResults.put(resultOption, true)
+  }
+
+  /** Conservatively protect a result with unknown ownership or reuse by 
another adaptive plan. */
+  private[adaptive] def markSharedStageResult(
+      resultOption: AtomicReference[Option[Any]], owner: 
AdaptiveSparkPlanExec): Unit = {
+    val originalOwner = stageResultOwners.get(resultOption)
+    if (originalOwner == null || (originalOwner ne owner)) {
+      markSharedStageResult(resultOption)
+    }
+  }
+
+  /** Return whether this result is conservatively protected from 
obsolete-stage cancellation. */
+  private[adaptive] def isSharedStageResult(resultOption: 
AtomicReference[Option[Any]]): Boolean = {
+    sharedStageResults.containsKey(resultOption)
+  }
+
   val shuffleIds: ConcurrentHashMap[Int, Boolean] = new ConcurrentHashMap[Int, 
Boolean]()
 }
 
diff --git 
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
index 9523ce897ecf..8e98d7785f6c 100644
--- 
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+++ 
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
@@ -19,7 +19,10 @@ package org.apache.spark.sql.execution.adaptive
 
 import java.io.File
 import java.net.URI
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
 
+import scala.concurrent.Future
 import scala.concurrent.duration._
 
 import org.apache.logging.log4j.Level
@@ -33,8 +36,8 @@ import org.apache.spark.sql.{DataFrame, Dataset, Row, 
SparkSession}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, 
AttributeReference, EqualTo, IsNull, Literal, Or, SortOrder}
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, 
EliminateLimits}
-import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, GlobalLimit, 
Join, JoinHint, LeafNode, LocalRelation, LogicalPlan, Statistics}
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, EmptyRelation, 
GlobalLimit, Join, JoinHint, LeafNode, LocalRelation, LogicalPlan, Statistics}
 import 
org.apache.spark.sql.catalyst.plans.physical.{CoalescedNullAwareHashPartitioning,
 SinglePartition}
 import org.apache.spark.sql.classic.Strategy
 import org.apache.spark.sql.execution._
@@ -60,8 +63,8 @@ import org.apache.spark.sql.test.SQLTestData.TestData
 import org.apache.spark.sql.types.{IntegerType, StructType}
 import org.apache.spark.sql.util.QueryExecutionListener
 import org.apache.spark.tags.SlowSQLTest
+import org.apache.spark.util.{SparkFatalException, ThreadUtils, Utils}
 import org.apache.spark.util.ArrayImplicits._
-import org.apache.spark.util.Utils
 
 /**
  * A leaf whose cost estimate under-counts its structural row bound. Used by 
SPARK-57956 to verify
@@ -373,6 +376,1061 @@ class AdaptiveQueryExecSuite
     }
   }
 
+  test("non-empty global aggregate stage eliminates conditionless semi joins") 
{
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+      val globalAggregate = spark.range(1).where("id < 0").repartition(2)
+        .agg(count("*").as("c"))
+      val df = testData.join(globalAggregate, Seq.empty[String], "left_semi")
+
+      val logicalJoin = df.queryExecution.optimizedPlan.collectFirst {
+        case join: Join => join
+      }.getOrElse(fail("expected a conditionless semi join before adaptive 
execution"))
+      assert(logicalJoin.joinType == LeftSemi)
+      assert(logicalJoin.condition.isEmpty)
+
+      val initialPlan = df.queryExecution.executedPlan
+        .asInstanceOf[AdaptiveSparkPlanExec].initialPlan
+      assert(findTopLevelBaseJoin(initialPlan).size == 1)
+
+      val aggregate = collect(initialPlan) {
+        case aggregate: BaseAggregateExec if 
aggregate.groupingExpressions.isEmpty => aggregate
+      }.headOption.getOrElse(fail("expected a global aggregate in the initial 
adaptive plan"))
+      val emptyStage = TestExchangeQueryStageExec(
+        0,
+        aggregate.child,
+        aggregate.child.canonicalized,
+        runtimeRowCount = Some(BigInt(0)))
+      emptyStage.resultOption.set(Some(()))
+      val aggregateStage = LogicalQueryStage(
+        logicalJoin.right, aggregate.withNewChildren(Seq(emptyStage)))
+      val rewrittenJoin = AQEPropagateEmptyRelation(logicalJoin.copy(right = 
aggregateStage))
+      assert(rewrittenJoin.fastEquals(logicalJoin.left), rewrittenJoin)
+
+      checkAnswer(df, testData.collect().toSeq)
+
+      val finalPlan = stripAQEPlan(df.queryExecution.executedPlan)
+      assert(findTopLevelBaseJoin(finalPlan).isEmpty, finalPlan)
+    }
+  }
+
+  test("global aggregate over empty filtered input preserves its single output 
row") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+      withTempView("empty_rows") {
+        spark.range(1).where("id < 0").createOrReplaceTempView("empty_rows")
+
+        val globalAggregate = sql(
+          "SELECT count(*) AS c FROM " +
+            "(SELECT /*+ REPARTITION(2) */ id FROM empty_rows ORDER BY id) 
sorted_empty_rows")
+        checkAnswer(globalAggregate, Row(0L))
+        val finalAggregatePlan = 
stripAQEPlan(globalAggregate.queryExecution.executedPlan)
+        assert(!finalAggregatePlan.isInstanceOf[EmptyRelationExec])
+
+        val df = sql(
+          "SELECT l.* FROM testData l LEFT ANTI JOIN " +
+            "(SELECT count(*) AS c FROM empty_rows HAVING count(*) = 0) r " +
+            "ON l.key = r.c + 1")
+
+        checkAnswer(df, testData.filter($"key" =!= 1).collect().toSeq)
+      }
+    }
+  }
+
+  test("sort and limit wrappers propagate only provably empty query stages") {
+    val getEstimatedRowCount =
+      PrivateMethod[Option[BigInt]](Symbol("getEstimatedRowCount"))
+    val scan = LocalTableScanExec(Nil, Nil, None)
+
+    val unknownStage = TestExchangeQueryStageExec(0, scan, scan)
+    val emptyStage = TestExchangeQueryStageExec(
+      1, scan, scan, runtimeRowCount = Some(BigInt(0)))
+    emptyStage.resultOption.set(Some(()))
+    val nonEmptyStage = TestExchangeQueryStageExec(
+      2, scan, scan, runtimeRowCount = Some(BigInt(10)))
+    nonEmptyStage.resultOption.set(Some(()))
+
+    def estimatedRowCount(plan: SparkPlan): Option[BigInt] =
+      AQEPropagateEmptyRelation.invokePrivate(getEstimatedRowCount(plan))
+
+    assert(estimatedRowCount(LocalLimitExec(0, 
unknownStage)).contains(BigInt(0)))
+    assert(estimatedRowCount(GlobalLimitExec(0, 
unknownStage)).contains(BigInt(0)))
+
+    assert(estimatedRowCount(LocalLimitExec(5, 
emptyStage)).contains(BigInt(0)))
+    assert(estimatedRowCount(GlobalLimitExec(5, 
emptyStage)).contains(BigInt(0)))
+    assert(estimatedRowCount(GlobalLimitExec(5, emptyStage, offset = 
2)).contains(BigInt(0)))
+    assert(estimatedRowCount(GlobalLimitExec(-1, emptyStage, offset = 
2)).contains(BigInt(0)))
+
+    assert(estimatedRowCount(LocalLimitExec(5, unknownStage)).isEmpty)
+    assert(estimatedRowCount(GlobalLimitExec(5, unknownStage)).isEmpty)
+    assert(estimatedRowCount(LocalLimitExec(1, nonEmptyStage)).isEmpty)
+    assert(estimatedRowCount(GlobalLimitExec(1, nonEmptyStage, offset = 
99)).isEmpty)
+
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.TOP_K_SORT_FALLBACK_THRESHOLD.key -> "1",
+      SQLConf.ORDERING_AWARE_LIMIT_OFFSET.key -> "true",
+      SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "false") {
+      val sorted = spark.range(0, 10, 1, numPartitions = 2)
+        .where("id < 0")
+        .orderBy($"id" % 8)
+        .limit(2)
+        .distinct()
+      val initialPlan = sorted.queryExecution.executedPlan
+        .asInstanceOf[AdaptiveSparkPlanExec].initialPlan
+      val (plannedSort, plannedShuffle) = initialPlan.collectFirst {
+        case GlobalLimitExec(_, sort @ SortExec(_, false, shuffle: 
ShuffleExchangeExec, _), _)
+            if sort.logicalLink.exists(logical =>
+              shuffle.logicalLink.exists(_ eq logical)) => (sort, shuffle)
+      }.getOrElse(fail("expected a logically linked sort above the 
middle-limit shuffle"))
+
+      val emptySortStage = TestExchangeQueryStageExec(
+        3,
+        plannedShuffle,
+        plannedShuffle.canonicalized,
+        runtimeRowCount = Some(BigInt(0)))
+      emptySortStage.resultOption.set(Some(()))
+      val sortedEmptyStage = plannedSort.withNewChildren(Seq(emptySortStage))
+
+      Seq(sortedEmptyStage, LocalLimitExec(5, sortedEmptyStage)).foreach { 
physicalPlan =>
+        val logicalStage = LogicalQueryStage(plannedSort.logicalLink.get, 
physicalPlan)
+        
assert(AQEPropagateEmptyRelation(logicalStage).isInstanceOf[EmptyRelation])
+      }
+
+      checkAnswer(sorted.toDF(), Seq.empty)
+    }
+  }
+
+  test("obsolete stage cancellation preserves shared reused exchange stages") {
+    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+      val df = spark.sql("SELECT * FROM testData join testData2 ON key = a")
+      val adaptivePlan = 
df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val replacementPlan = LocalTableScanExec(Nil, Nil, None)
+
+      val sharedStage = TestExchangeQueryStageExec(
+        0, LocalTableScanExec(Nil, Nil, None), LocalTableScanExec(Nil, Nil, 
None))
+      adaptivePlan.context.markSharedStageResult(sharedStage.resultOption)
+
+      val sharedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(sharedStage)))
+      assert(sharedCancelledIds.isEmpty)
+      assert(!sharedStage.cancelled)
+
+      val materializedStage = TestExchangeQueryStageExec(
+        1, LocalTableScanExec(Nil, Nil, None), LocalTableScanExec(Nil, Nil, 
None))
+      materializedStage.resultOption.set(Some(()))
+      val materializedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(materializedStage)))
+      assert(materializedCancelledIds.isEmpty)
+      assert(!materializedStage.cancelled)
+
+      val retainedStage = TestExchangeQueryStageExec(
+        2, LocalTableScanExec(Nil, Nil, None), LocalTableScanExec(Nil, Nil, 
None))
+      val retainedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(retainedStage, Seq(retainedStage)))
+      assert(retainedCancelledIds.isEmpty)
+      assert(!retainedStage.cancelled)
+
+      val reusedStage = TestExchangeQueryStageExec(
+        3, LocalTableScanExec(Nil, Nil, None), LocalTableScanExec(Nil, Nil, 
None))
+      val reusedReplacementStage = reusedStage.newReuseInstance(4, 
reusedStage.output)
+      val reusedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(reusedReplacementStage, Seq(reusedStage)))
+      assert(reusedCancelledIds.isEmpty)
+      assert(!reusedStage.cancelled)
+
+      val unsharedStage = TestExchangeQueryStageExec(
+        5, LocalTableScanExec(Nil, Nil, None), LocalTableScanExec(Nil, Nil, 
None))
+      adaptivePlan.context.stageCache.put(unsharedStage.plan.canonicalized, 
unsharedStage)
+      val unsharedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(unsharedStage)))
+      assert(unsharedCancelledIds == Seq(unsharedStage.id))
+      assert(unsharedStage.cancelled)
+      
assert(!adaptivePlan.context.stageCache.contains(unsharedStage.plan.canonicalized))
+
+      val failedStage = TestExchangeQueryStageExec(
+        6,
+        LocalTableScanExec(Nil, Nil, None),
+        LocalTableScanExec(Nil, Nil, None),
+        cancelFailure = Some(new IllegalStateException("test stage 
cancellation failed")))
+      adaptivePlan.context.stageCache.put(failedStage.plan.canonicalized, 
failedStage)
+      val failedCancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(failedStage)))
+      assert(failedCancelledIds.isEmpty)
+      assert(!failedStage.cancelled)
+      
assert(adaptivePlan.context.stageCache.get(failedStage.plan.canonicalized)
+        .exists(_ eq failedStage))
+    }
+  }
+
+  test("obsolete same-plan reused exchange aliases are cancelled only when all 
are removed") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val adaptivePlan = spark.sql("SELECT * FROM testData JOIN testData2 ON 
key = a")
+        .queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case shuffle: ShuffleExchangeExec => shuffle
+      }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      def acquireStage(): ExchangeQueryStageExec = {
+        adaptivePlan.invokePrivate(createNonResultQueryStages(exchange))
+          
.asInstanceOf[Product].productElement(0).asInstanceOf[ExchangeQueryStageExec]
+      }
+
+      val originalStage = acquireStage()
+      val reusedStage = acquireStage()
+      assert(originalStage.id != reusedStage.id)
+      assert(originalStage.resultOption.eq(reusedStage.resultOption))
+
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val candidateStages = Seq(originalStage, reusedStage)
+
+      // Retaining either alias keeps the shared physical exchange alive.
+      assert(adaptivePlan.invokePrivate(
+        cancelObsoleteStages(reusedStage, candidateStages)).isEmpty)
+      
assert(adaptivePlan.context.stageCache.get(originalStage.plan.canonicalized)
+        .exists(_.resultOption.eq(originalStage.resultOption)))
+
+      val emptyReplacement = LocalTableScanExec(Nil, Nil, None)
+      val cancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(emptyReplacement, candidateStages))
+      assert(cancelledIds.toSet == Set(originalStage.id, reusedStage.id))
+      assert(cancelledIds.distinct.size == candidateStages.size)
+      
assert(!adaptivePlan.context.stageCache.contains(originalStage.plan.canonicalized))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val cancellationFailure = new IllegalStateException("obsolete reused 
exchange was cancelled")
+      candidateStages.foreach { stage =>
+        assert(adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+          stage, cancellationFailure, cancelledIds.toSet)))
+        assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+          stage, new SparkFatalException(new OutOfMemoryError("fatal obsolete 
alias")),
+          cancelledIds.toSet)))
+      }
+
+      val unrelatedStage = TestExchangeQueryStageExec(
+        412, exchange, exchange.canonicalized)
+      assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        unrelatedStage, cancellationFailure, cancelledIds.toSet)))
+    }
+  }
+
+  test("obsolete exchange cancellation preserves reuse by another adaptive 
subquery plan") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val adaptivePlan = spark.sql("SELECT * FROM testData JOIN testData2 ON 
key = a")
+        .queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case shuffle: ShuffleExchangeExec => shuffle
+      }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      def acquireStage(plan: AdaptiveSparkPlanExec): ExchangeQueryStageExec = {
+        plan.invokePrivate(createNonResultQueryStages(exchange))
+          
.asInstanceOf[Product].productElement(0).asInstanceOf[ExchangeQueryStageExec]
+      }
+
+      val originalStage = acquireStage(adaptivePlan)
+      val samePlanAlias = acquireStage(adaptivePlan)
+      val subqueryPlan = adaptivePlan.copy(isSubquery = true)
+      assert(subqueryPlan ne adaptivePlan)
+      assert(subqueryPlan == adaptivePlan)
+      assert(subqueryPlan.context eq adaptivePlan.context)
+
+      val subqueryAlias = acquireStage(subqueryPlan)
+      assert(originalStage.resultOption.eq(samePlanAlias.resultOption))
+      assert(originalStage.resultOption.eq(subqueryAlias.resultOption))
+
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val emptyReplacement = LocalTableScanExec(Nil, Nil, None)
+      assert(adaptivePlan.invokePrivate(cancelObsoleteStages(
+        emptyReplacement, Seq(originalStage, samePlanAlias))).isEmpty)
+      
assert(adaptivePlan.context.stageCache.get(originalStage.plan.canonicalized)
+        .exists(_.resultOption.eq(subqueryAlias.resultOption)))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val requiredStageFailure = new IllegalStateException("required subquery 
exchange failed")
+      assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        originalStage, requiredStageFailure, Set.empty[Int])))
+      assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        samePlanAlias, requiredStageFailure, Set.empty[Int])))
+    }
+  }
+
+  test("failed obsolete exchange cancellation is not retried after same-plan 
reuse") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val adaptivePlan = spark.sql("SELECT * FROM testData JOIN testData2 ON 
key = a")
+        .queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case shuffle: ShuffleExchangeExec => shuffle
+      }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+
+      val cancellationAttempts = new AtomicInteger()
+      val originalStage = TestExchangeQueryStageExec(
+        413,
+        exchange,
+        exchange.canonicalized,
+        cancelCallback = Some(() => {
+          if (cancellationAttempts.incrementAndGet() == 1) {
+            throw new IllegalStateException("first stage cancellation failed")
+          }
+        }))
+      adaptivePlan.context.registerStageOwner(originalStage.resultOption, 
adaptivePlan)
+      adaptivePlan.context.stageCache.put(exchange.canonicalized, 
originalStage)
+
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val emptyReplacement = LocalTableScanExec(Nil, Nil, None)
+      assert(adaptivePlan.invokePrivate(
+        cancelObsoleteStages(emptyReplacement, Seq(originalStage))).isEmpty)
+      assert(cancellationAttempts.get() == 1)
+      assert(!originalStage.cancelled)
+      
assert(adaptivePlan.context.stageCache.get(exchange.canonicalized).contains(originalStage))
+
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      val reusedStage = 
adaptivePlan.invokePrivate(createNonResultQueryStages(exchange))
+        
.asInstanceOf[Product].productElement(0).asInstanceOf[ExchangeQueryStageExec]
+      assert(originalStage.id != reusedStage.id)
+      assert(originalStage.resultOption.eq(reusedStage.resultOption))
+      
assert(!adaptivePlan.context.isSharedStageResult(originalStage.resultOption))
+
+      assert(adaptivePlan.invokePrivate(
+        cancelObsoleteStages(emptyReplacement, Seq(reusedStage))).isEmpty)
+      assert(cancellationAttempts.get() == 1,
+        "a failed exchange cancellation must never be retried for another 
local alias")
+      assert(!originalStage.cancelled)
+      
assert(adaptivePlan.context.stageCache.get(exchange.canonicalized).contains(originalStage))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val fatalFailure = new SparkFatalException(new OutOfMemoryError("fatal 
failed cancellation"))
+      Seq(originalStage, reusedStage).foreach { stage =>
+        assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+          stage, fatalFailure, Set.empty[Int])))
+      }
+
+      val obsoleteFailure = new IllegalStateException("uncancelled obsolete 
exchange failed")
+      Seq(originalStage, reusedStage).foreach { stage =>
+        assert(adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+          stage, obsoleteFailure, Set.empty[Int])))
+      }
+      assert(cancellationAttempts.get() == 1)
+    }
+  }
+
+  test("obsolete reused exchange cancellation includes aliases dropped by an 
earlier plan") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val adaptivePlan = spark.sql("SELECT * FROM testData JOIN testData2 ON 
key = a")
+        .queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case shuffle: ShuffleExchangeExec => shuffle
+      }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      def acquireStage(): ExchangeQueryStageExec = {
+        adaptivePlan.invokePrivate(createNonResultQueryStages(exchange))
+          
.asInstanceOf[Product].productElement(0).asInstanceOf[ExchangeQueryStageExec]
+      }
+
+      val originalStage = acquireStage()
+      val previouslyDroppedAlias = acquireStage()
+      val lastRemainingAlias = acquireStage()
+      val allAliases = Seq(originalStage, previouslyDroppedAlias, 
lastRemainingAlias)
+      assert(allAliases.map(_.id).distinct.size == allAliases.size)
+      assert(allAliases.forall(_.resultOption.eq(originalStage.resultOption)))
+
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      assert(adaptivePlan.invokePrivate(
+        cancelObsoleteStages(lastRemainingAlias, allAliases)).isEmpty)
+
+      val emptyReplacement = LocalTableScanExec(Nil, Nil, None)
+      val cancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(emptyReplacement, Seq(lastRemainingAlias)))
+      assert(cancelledIds.toSet == allAliases.map(_.id).toSet)
+      assert(cancelledIds.distinct.size == allAliases.size)
+      
assert(!adaptivePlan.context.stageCache.contains(originalStage.plan.canonicalized))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val cancellationFailure = new IllegalStateException("previously dropped 
exchange failed")
+      allAliases.foreach { alias =>
+        assert(adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+          alias, cancellationFailure, cancelledIds.toSet)))
+      }
+    }
+  }
+
+  test("cancelling a submitted reused shuffle alias preserves its underlying 
shuffle cleanup ID") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+      val owner = org.apache.spark.sql.classic.Dataset.ofRows(
+        spark, spark.range(16).repartition(2).logicalPlan, DoNotCleanup)
+      val adaptivePlan = 
owner.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      assert(adaptivePlan.context.qe.shuffleCleanupMode == DoNotCleanup)
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case shuffle: ShuffleExchangeExec => shuffle
+      }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      def acquireStage(): ShuffleQueryStageExec = {
+        adaptivePlan.invokePrivate(createNonResultQueryStages(exchange))
+          
.asInstanceOf[Product].productElement(0).asInstanceOf[ShuffleQueryStageExec]
+      }
+
+      val originalStage = acquireStage()
+      val reusedStage = acquireStage()
+      assert(reusedStage.shuffle eq originalStage.shuffle)
+      assert(reusedStage.plan.isInstanceOf[ReusedExchangeExec])
+      ThreadUtils.awaitResult(originalStage.materialize(), 30.seconds)
+      assert(originalStage.shuffle.futureAction.get().isDefined)
+      assert(!originalStage.isMaterialized)
+
+      val shuffleId = originalStage.shuffle.shuffleId
+      adaptivePlan.context.shuffleIds.remove(shuffleId)
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val emptyReplacement = LocalTableScanExec(Nil, Nil, None)
+      val cancelledIds = adaptivePlan.invokePrivate(
+        cancelObsoleteStages(emptyReplacement, Seq(reusedStage)))
+
+      assert(cancelledIds.toSet == Set(originalStage.id, reusedStage.id))
+      assert(adaptivePlan.context.shuffleIds.containsKey(shuffleId),
+        "a reused exchange leaf must not hide its submitted underlying shuffle 
from cleanup")
+      
assert(!adaptivePlan.context.stageCache.contains(originalStage.plan.canonicalized))
+    }
+  }
+
+  test("obsolete stage cancellation includes stages from previous adaptive 
plan adoptions") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val adaptivePlan = spark.sql("SELECT * FROM testData JOIN testData2 ON 
key = a")
+        .queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchanges = adaptivePlan.initialPlan.collect {
+        case exchange: ShuffleExchangeExec => exchange
+      }
+      assert(exchanges.size >= 2, "expected two shuffle exchanges in the 
initial adaptive plan")
+      val previousExchange = exchanges.head
+      val latestExchange = exchanges.find(exchange =>
+        !exchange.canonicalized.fastEquals(previousExchange.canonicalized))
+        .getOrElse(fail("expected independently cached shuffle exchanges"))
+
+      val previouslyRetainedStage = TestExchangeQueryStageExec(
+        410, previousExchange, previousExchange.canonicalized)
+      val latestStage = TestExchangeQueryStageExec(
+        411, latestExchange, latestExchange.canonicalized)
+      val oldPhysicalPlan = UnionExec(Seq(previouslyRetainedStage, 
latestStage))
+      val latestStages = Seq(latestStage)
+      adaptivePlan.context.stageCache.put(previousExchange.canonicalized, 
previouslyRetainedStage)
+      adaptivePlan.context.stageCache.put(latestExchange.canonicalized, 
latestStage)
+
+      val obsoleteStageCandidates =
+        PrivateMethod[Seq[QueryStageExec]](Symbol("obsoleteStageCandidates"))
+      val candidates = adaptivePlan.invokePrivate(
+        obsoleteStageCandidates(oldPhysicalPlan, latestStages))
+
+      assert(!latestStages.exists(_.id == previouslyRetainedStage.id))
+      assert(candidates.map(_.id).toSet == Set(previouslyRetainedStage.id, 
latestStage.id))
+      assert(candidates.count(_.id == latestStage.id) == 1)
+
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val cancelledIds = 
adaptivePlan.invokePrivate(cancelObsoleteStages(latestStage, candidates))
+      assert(cancelledIds == Seq(previouslyRetainedStage.id))
+      assert(previouslyRetainedStage.cancelled)
+      assert(!latestStage.cancelled)
+      
assert(!adaptivePlan.context.stageCache.contains(previousExchange.canonicalized))
+      assert(adaptivePlan.context.stageCache.get(latestExchange.canonicalized)
+        .exists(_ eq latestStage))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val stageFailure = new IllegalStateException("obsolete stage failed 
after cancellation")
+      assert(adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        previouslyRetainedStage, stageFailure, cancelledIds.toSet)))
+      assert(!adaptivePlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        latestStage, stageFailure, cancelledIds.toSet)))
+    }
+  }
+
+  test("obsolete broadcast stages are not cancelled or initialized") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      val df = spark.sql(
+        "SELECT /*+ BROADCAST(r) */ * FROM testData l JOIN testData2 r ON 
l.key = r.a")
+      val adaptivePlan = 
df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val exchange = adaptivePlan.initialPlan.collectFirst {
+        case broadcast: BroadcastExchangeExec => broadcast
+      }.getOrElse(fail("expected an uninitialized broadcast exchange"))
+
+      val stage = BroadcastQueryStageExec(400, exchange, 
exchange.canonicalized)
+      adaptivePlan.context.stageCache.put(exchange.canonicalized, stage)
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val replacementPlan = LocalTableScanExec(Nil, Nil, None)
+
+      assert(adaptivePlan.invokePrivate(cancelObsoleteStages(replacementPlan, 
Seq(stage))).isEmpty)
+      assert(stage.resultOption.get().isEmpty)
+      
assert(adaptivePlan.context.stageCache.get(exchange.canonicalized).exists(_ eq 
stage))
+    }
+  }
+
+  test("fatal obsolete stage failures are never ignored") {
+    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+      val df = spark.sql("SELECT * FROM testData JOIN testData2 ON key = a")
+      val adaptivePlan = 
df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val scan = LocalTableScanExec(Nil, Nil, None)
+      val stage = TestExchangeQueryStageExec(401, scan, scan)
+      val cancelledStageIds = Set(stage.id)
+      val wrappedFatal = new SparkFatalException(new OutOfMemoryError("fatal 
stage failure"))
+      val rawFatal = new OutOfMemoryError("fatal stage failure")
+      val cancellationFailure = new 
java.util.concurrent.CancellationException("stage cancelled")
+
+      assert(!adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(stage, wrappedFatal, 
cancelledStageIds)))
+      assert(!adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(stage, rawFatal, cancelledStageIds)))
+      assert(adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(stage, cancellationFailure, 
cancelledStageIds)))
+      assert(!adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(stage, cancellationFailure, 
Set.empty[Int])))
+
+      val failedStage = TestExchangeQueryStageExec(
+        402,
+        scan,
+        scan,
+        cancelFailure = Some(new IllegalStateException("stage cancellation 
failed")))
+      adaptivePlan.context.stageCache.put(scan.canonicalized, failedStage)
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      assert(adaptivePlan.invokePrivate(cancelObsoleteStages(scan, 
Seq(failedStage))).isEmpty)
+
+      failedStage.error.set(Some(wrappedFatal))
+      assert(!adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(failedStage, wrappedFatal, 
Set.empty[Int])))
+      assert(!adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(failedStage, rawFatal, 
Set.empty[Int])))
+      assert(adaptivePlan.context.stageCache.get(scan.canonicalized).exists(_ 
eq failedStage))
+
+      failedStage.error.set(Some(cancellationFailure))
+      assert(adaptivePlan.invokePrivate(
+        shouldIgnoreObsoleteStageFailure(failedStage, cancellationFailure, 
Set.empty[Int])))
+      assert(!adaptivePlan.context.stageCache.contains(scan.canonicalized))
+    }
+  }
+
+  test("obsolete shuffle stage cancellation records only submitted shuffles 
for cleanup") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val replacementPlan = LocalTableScanExec(Nil, Nil, None)
+
+      val unsubmitted = spark.range(8).repartition(2)
+      val unsubmittedPlan =
+        
unsubmitted.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val unsubmittedExchange = unsubmittedPlan.initialPlan.collectFirst {
+        case exchange: ShuffleExchangeExec => exchange
+      }.getOrElse(fail("expected an unsubmitted shuffle exchange"))
+      val unsubmittedStage = TestExchangeQueryStageExec(
+        200, unsubmittedExchange, unsubmittedExchange.canonicalized)
+      
unsubmittedPlan.context.stageCache.put(unsubmittedExchange.canonicalized, 
unsubmittedStage)
+
+      assert(unsubmittedExchange.futureAction.get().isEmpty)
+      assert(unsubmittedPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(unsubmittedStage))) == 
Seq(unsubmittedStage.id))
+      assert(unsubmittedPlan.context.shuffleIds.isEmpty)
+      assert(unsubmittedExchange.futureAction.get().isEmpty)
+
+      val submitted = spark.range(8).repartition(2)
+      submitted.collect()
+      val submittedPlan = 
submitted.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val submittedExchange = collect(submittedPlan) {
+        case stage: ShuffleQueryStageExec if 
stage.shuffle.futureAction.get().isDefined =>
+          stage.shuffle
+      }.headOption.getOrElse(fail("expected a submitted shuffle exchange"))
+      val submittedShuffleId = submittedExchange.shuffleId
+      submittedPlan.context.shuffleIds.remove(submittedShuffleId)
+      val submittedStage = TestExchangeQueryStageExec(
+        201, submittedExchange, submittedExchange.canonicalized)
+      submittedPlan.context.stageCache.put(submittedExchange.canonicalized, 
submittedStage)
+
+      assert(submittedPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(submittedStage))) == 
Seq(submittedStage.id))
+      assert(submittedPlan.context.shuffleIds.containsKey(submittedShuffleId))
+
+      submittedPlan.context.shuffleIds.remove(submittedShuffleId)
+      val failedSubmittedStage = TestExchangeQueryStageExec(
+        202,
+        submittedExchange,
+        submittedExchange.canonicalized,
+        cancelFailure = Some(new IllegalStateException("submitted stage 
cancellation failed")))
+      submittedPlan.context.stageCache.put(submittedExchange.canonicalized, 
failedSubmittedStage)
+
+      assert(submittedPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, 
Seq(failedSubmittedStage))).isEmpty)
+      assert(submittedPlan.context.shuffleIds.containsKey(submittedShuffleId))
+      
assert(submittedPlan.context.stageCache.get(submittedExchange.canonicalized)
+        .exists(_ eq failedSubmittedStage))
+
+      val delegatedExchange = org.apache.spark.sql.MyShuffleExchangeExec(
+        submittedExchange.asInstanceOf[ShuffleExchangeExec])
+      val delegatedStage = ShuffleQueryStageExec(
+        203, delegatedExchange, delegatedExchange.canonicalized)
+      submittedPlan.context.stageCache.put(delegatedExchange.canonicalized, 
delegatedStage)
+
+      assert(submittedExchange.futureAction.get().isDefined)
+      assert(delegatedExchange.futureAction.get().isEmpty)
+      assert(submittedPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(delegatedStage))).isEmpty)
+      
assert(submittedPlan.context.stageCache.get(delegatedExchange.canonicalized)
+        .exists(_ eq delegatedStage))
+
+      submittedPlan.context.shuffleIds.remove(submittedShuffleId)
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val delegatedFailure = new IllegalStateException("opaque delegated 
shuffle failed")
+      delegatedStage.error.set(Some(delegatedFailure))
+      assert(!submittedPlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        delegatedStage, delegatedFailure, Set.empty[Int])))
+      assert(!submittedPlan.context.shuffleIds.containsKey(submittedShuffleId))
+      assert(submittedExchange.futureAction.get().exists(!_.isCancelled))
+      
assert(submittedPlan.context.stageCache.get(delegatedExchange.canonicalized)
+        .exists(_ eq delegatedStage))
+    }
+  }
+
+  test("obsolete submitted shuffle stages are not cancelled before 
shuffle-file cleanup") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val replacementPlan = LocalTableScanExec(Nil, Nil, None)
+
+      val submitted = spark.range(8).repartition(2)
+      submitted.collect()
+      val submittedPlan = 
submitted.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      val submittedExchange = collect(submittedPlan) {
+        case stage: ShuffleQueryStageExec if 
stage.shuffle.futureAction.get().isDefined =>
+          stage.shuffle
+      }.headOption.getOrElse(fail("expected a submitted shuffle exchange"))
+      val submittedAction = submittedExchange.futureAction.get().get
+
+      val cleanup = org.apache.spark.sql.classic.Dataset.ofRows(
+        spark, spark.range(16).repartition(2).logicalPlan, RemoveShuffleFiles)
+      val cleanupPlan = 
cleanup.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+      assert(cleanupPlan.context.qe.shuffleCleanupMode == RemoveShuffleFiles)
+
+      val submittedStage = ShuffleQueryStageExec(
+        300, submittedExchange, submittedExchange.canonicalized)
+      cleanupPlan.context.stageCache.put(submittedExchange.canonicalized, 
submittedStage)
+
+      assert(!submittedAction.isCancelled)
+      assert(cleanupPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(submittedStage))).isEmpty)
+      assert(!submittedAction.isCancelled)
+      
assert(cleanupPlan.context.stageCache.get(submittedExchange.canonicalized)
+        .exists(_ eq submittedStage))
+
+      val shouldIgnoreObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("shouldIgnoreObsoleteStageFailure"))
+      val fatalFailure = new SparkFatalException(new OutOfMemoryError("fatal 
shuffle failure"))
+      assert(!cleanupPlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        submittedStage, fatalFailure, Set.empty[Int])))
+
+      val stageFailure = new IllegalStateException("obsolete submitted shuffle 
failed")
+      submittedStage.error.set(Some(stageFailure))
+      assert(cleanupPlan.invokePrivate(shouldIgnoreObsoleteStageFailure(
+        submittedStage, stageFailure, Set.empty[Int])))
+      
assert(cleanupPlan.context.shuffleIds.containsKey(submittedExchange.shuffleId))
+      
assert(!cleanupPlan.context.stageCache.contains(submittedExchange.canonicalized))
+      assert(!submittedAction.isCancelled)
+
+      val unsubmittedExchange = cleanupPlan.initialPlan.collectFirst {
+        case exchange: ShuffleExchangeExec => exchange
+      }.getOrElse(fail("expected an unsubmitted shuffle exchange"))
+      val unsubmittedStage = ShuffleQueryStageExec(
+        301, unsubmittedExchange, unsubmittedExchange.canonicalized)
+      cleanupPlan.context.stageCache.put(unsubmittedExchange.canonicalized, 
unsubmittedStage)
+
+      assert(unsubmittedExchange.futureAction.get().isEmpty)
+      assert(cleanupPlan.invokePrivate(
+        cancelObsoleteStages(replacementPlan, Seq(unsubmittedStage))) == 
Seq(unsubmittedStage.id))
+      assert(unsubmittedExchange.futureAction.get().isEmpty)
+      
assert(!cleanupPlan.context.stageCache.contains(unsubmittedExchange.canonicalized))
+    }
+  }
+
+  test("concurrent exchange reuse and obsolete cancellation coordinate by 
stage") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
+      val createNonResultQueryStages =
+        PrivateMethod[Any](Symbol("createNonResultQueryStages"))
+      val cancelObsoleteStages =
+        PrivateMethod[Seq[Int]](Symbol("cancelObsoleteStages"))
+      val ignoreFailedObsoleteStageFailure =
+        PrivateMethod[Boolean](Symbol("ignoreFailedObsoleteStageFailure"))
+      val replacementPlan = LocalTableScanExec(Nil, Nil, None)
+
+      def newPlanAndExchange(): (AdaptiveSparkPlanExec, ShuffleExchangeExec) = 
{
+        val df = spark.sql("SELECT * FROM testData JOIN testData2 ON key = a")
+        val adaptivePlan = 
df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
+        val exchange = adaptivePlan.initialPlan.collectFirst {
+          case shuffle: ShuffleExchangeExec => shuffle
+        }.getOrElse(fail("expected a shuffle exchange in the initial adaptive 
plan"))
+        (adaptivePlan, exchange)
+      }
+
+      def acquireStage(
+          adaptivePlan: AdaptiveSparkPlanExec,
+          exchange: ShuffleExchangeExec): ExchangeQueryStageExec = {
+        adaptivePlan.invokePrivate(createNonResultQueryStages(exchange))
+          
.asInstanceOf[Product].productElement(0).asInstanceOf[ExchangeQueryStageExec]
+      }
+
+      type LifecycleWorker = (Thread, AtomicReference[Throwable])
+
+      def startWorker(name: String)(body: => Unit): LifecycleWorker = {
+        val started = new CountDownLatch(1)
+        val failure = new AtomicReference[Throwable]()
+        val worker = new Thread(name) {
+          override def run(): Unit = {
+            started.countDown()
+            try {
+              spark.withActive {
+                body
+              }
+            } catch {
+              case error: Throwable => failure.set(error)
+            }
+          }
+        }
+        worker.setDaemon(true)
+        worker.start()
+        assert(started.await(30, TimeUnit.SECONDS), s"$name did not start")
+        (worker, failure)
+      }
+
+      def waitForWorker(worker: LifecycleWorker): Unit = {
+        worker._1.join(TimeUnit.SECONDS.toMillis(30))
+      }
+
+      def checkWorker(worker: LifecycleWorker): Unit = {
+        assert(!worker._1.isAlive, s"${worker._1.getName} did not finish")
+        Option(worker._2.get()).foreach(throw _)
+      }
+
+      {
+        val (adaptivePlan, exchange) = newPlanAndExchange()
+        val reuseEntered = new CountDownLatch(1)
+        val releaseReuse = new CountDownLatch(1)
+        val existingStage = TestExchangeQueryStageExec(
+          100,
+          exchange,
+          exchange.canonicalized,
+          reuseCallback = Some(() => {
+            reuseEntered.countDown()
+            assert(releaseReuse.await(30, TimeUnit.SECONDS), "stage reuse was 
not released")
+          }))
+        adaptivePlan.context.stageCache.put(exchange.canonicalized, 
existingStage)
+
+        val acquiredStage = new AtomicReference[ExchangeQueryStageExec]()
+        val cancelledIds = new AtomicReference[Seq[Int]]()
+        val reuseWorker = startWorker("aqe-stage-reuse-first") {
+          acquiredStage.set(acquireStage(adaptivePlan, exchange))
+        }
+        var cancellationWorker: Option[LifecycleWorker] = None
+
+        try {
+          assert(reuseEntered.await(30, TimeUnit.SECONDS), "exchange reuse did 
not start")
+          cancellationWorker = 
Some(startWorker("aqe-stage-cancellation-after-reuse") {
+            cancelledIds.set(adaptivePlan.invokePrivate(
+              cancelObsoleteStages(replacementPlan, Seq(existingStage))))
+          })
+          eventually(timeout(10.seconds), interval(10.milliseconds)) {
+            assert(cancellationWorker.get._1.getState == Thread.State.BLOCKED)
+          }
+        } finally {
+          releaseReuse.countDown()
+          waitForWorker(reuseWorker)
+          cancellationWorker.foreach(waitForWorker)
+        }
+
+        checkWorker(reuseWorker)
+        cancellationWorker.foreach(checkWorker)
+        assert(cancelledIds.get().isEmpty)
+        assert(!existingStage.cancelled)
+        assert(acquiredStage.get().resultOption.eq(existingStage.resultOption))
+        
assert(adaptivePlan.context.isSharedStageResult(existingStage.resultOption))
+        assert(adaptivePlan.context.stageCache.get(exchange.canonicalized)
+          .exists(_ eq existingStage))
+      }
+
+      {
+        val (adaptivePlan, exchange) = newPlanAndExchange()
+        val unrelatedExchange = adaptivePlan.initialPlan.collectFirst {
+          case candidate: ShuffleExchangeExec
+              if !candidate.canonicalized.fastEquals(exchange.canonicalized) 
=> candidate
+        }.getOrElse(fail("expected an unrelated shuffle exchange in the 
initial adaptive plan"))
+        val cancellationEntered = new CountDownLatch(1)
+        val releaseCancellation = new CountDownLatch(1)
+        val obsoleteStage = TestExchangeQueryStageExec(
+          101,
+          exchange,
+          exchange.canonicalized,
+          cancelCallback = Some(() => {
+            cancellationEntered.countDown()
+            assert(releaseCancellation.await(30, TimeUnit.SECONDS),
+              "stage cancellation was not released")
+          }))
+        adaptivePlan.context.stageCache.put(exchange.canonicalized, 
obsoleteStage)
+
+        val acquiredStage = new AtomicReference[ExchangeQueryStageExec]()
+        val unrelatedStage = new AtomicReference[ExchangeQueryStageExec]()
+        val unrelatedAcquired = new CountDownLatch(1)
+        val cancelledIds = new AtomicReference[Seq[Int]]()
+        val cancellationWorker = startWorker("aqe-stage-cancellation-first") {
+          cancelledIds.set(adaptivePlan.invokePrivate(
+            cancelObsoleteStages(replacementPlan, Seq(obsoleteStage))))
+        }
+        var reuseWorker: Option[LifecycleWorker] = None
+        var unrelatedWorker: Option[LifecycleWorker] = None
+
+        try {
+          assert(cancellationEntered.await(30, TimeUnit.SECONDS),
+            "obsolete stage cancellation did not start")
+          reuseWorker = Some(startWorker("aqe-stage-reuse-after-cancellation") 
{
+            acquiredStage.set(acquireStage(adaptivePlan, exchange))
+          })
+          eventually(timeout(10.seconds), interval(10.milliseconds)) {
+            assert(Set(Thread.State.BLOCKED, Thread.State.WAITING, 
Thread.State.TIMED_WAITING)
+              .contains(reuseWorker.get._1.getState))
+            assert(acquiredStage.get() == null)
+          }
+
+          unrelatedWorker = 
Some(startWorker("aqe-unrelated-stage-during-cancellation") {
+            unrelatedStage.set(acquireStage(adaptivePlan, unrelatedExchange))
+            unrelatedAcquired.countDown()
+          })
+          assert(unrelatedAcquired.await(10, TimeUnit.SECONDS),
+            "an unrelated exchange was blocked while stage cancellation was 
running")
+          
assert(unrelatedStage.get().resultOption.ne(obsoleteStage.resultOption))
+          assert(acquiredStage.get() == null)
+        } finally {
+          releaseCancellation.countDown()
+          waitForWorker(cancellationWorker)
+          reuseWorker.foreach(waitForWorker)
+          unrelatedWorker.foreach(waitForWorker)
+        }
+
+        checkWorker(cancellationWorker)
+        reuseWorker.foreach(checkWorker)
+        unrelatedWorker.foreach(checkWorker)
+        assert(cancelledIds.get() == Seq(obsoleteStage.id))
+        assert(obsoleteStage.cancelled)
+        assert(acquiredStage.get().resultOption.ne(obsoleteStage.resultOption))
+        
assert(adaptivePlan.context.stageCache.get(acquiredStage.get().plan.canonicalized)
+          .exists(_ eq acquiredStage.get()))
+        
assert(adaptivePlan.context.stageCache.get(unrelatedStage.get().plan.canonicalized)
+          .exists(_ eq unrelatedStage.get()))
+      }
+
+      {
+        val (adaptivePlan, obsoleteExchange) = newPlanAndExchange()
+        val unrelatedExchange = adaptivePlan.initialPlan.collectFirst {
+          case exchange: ShuffleExchangeExec
+              if 
!exchange.canonicalized.fastEquals(obsoleteExchange.canonicalized) => exchange
+        }.getOrElse(fail("expected an unrelated shuffle exchange in the 
initial adaptive plan"))
+        val obsoleteStage = ShuffleQueryStageExec(
+          104, obsoleteExchange, obsoleteExchange.canonicalized)
+        adaptivePlan.context.stageCache.put(obsoleteExchange.canonicalized, 
obsoleteStage)
+
+        val monitorEntered = new CountDownLatch(1)
+        val releaseMonitor = new CountDownLatch(1)
+        val monitorWorker = startWorker("aqe-obsolete-shuffle-monitor-held") {
+          obsoleteExchange.synchronized {
+            monitorEntered.countDown()
+            assert(releaseMonitor.await(30, TimeUnit.SECONDS),
+              "obsolete shuffle monitor was not released")
+          }
+        }
+
+        val cancelledIds = new AtomicReference[Seq[Int]]()
+        val acquiredStage = new AtomicReference[ExchangeQueryStageExec]()
+        val unrelatedAcquired = new CountDownLatch(1)
+        var cancellationWorker: Option[LifecycleWorker] = None
+        var unrelatedWorker: Option[LifecycleWorker] = None
+
+        try {
+          assert(monitorEntered.await(30, TimeUnit.SECONDS),
+            "obsolete shuffle monitor was not acquired")
+          cancellationWorker = 
Some(startWorker("aqe-shuffle-cancellation-waits-for-monitor") {
+            cancelledIds.set(adaptivePlan.invokePrivate(
+              cancelObsoleteStages(replacementPlan, Seq(obsoleteStage))))
+          })
+          eventually(timeout(10.seconds), interval(10.milliseconds)) {
+            assert(cancellationWorker.get._1.getState == Thread.State.BLOCKED)
+          }
+
+          unrelatedWorker = 
Some(startWorker("aqe-unrelated-stage-acquisition") {
+            acquiredStage.set(acquireStage(adaptivePlan, unrelatedExchange))
+            unrelatedAcquired.countDown()
+          })
+          assert(unrelatedAcquired.await(10, TimeUnit.SECONDS),
+            "an unrelated exchange was blocked by obsolete shuffle 
cancellation")
+          
assert(acquiredStage.get().resultOption.ne(obsoleteStage.resultOption))
+          
assert(adaptivePlan.context.stageCache.get(acquiredStage.get().plan.canonicalized)
+            .exists(_ eq acquiredStage.get()))
+        } finally {
+          releaseMonitor.countDown()
+          waitForWorker(monitorWorker)
+          cancellationWorker.foreach(waitForWorker)
+          unrelatedWorker.foreach(waitForWorker)
+        }
+
+        checkWorker(monitorWorker)
+        cancellationWorker.foreach(checkWorker)
+        unrelatedWorker.foreach(checkWorker)
+        assert(cancelledIds.get() == Seq(obsoleteStage.id))
+        
assert(!adaptivePlan.context.stageCache.contains(obsoleteExchange.canonicalized))
+      }
+
+      {
+        val (adaptivePlan, exchange) = newPlanAndExchange()
+        val cancellationEntered = new CountDownLatch(1)
+        val releaseCancellation = new CountDownLatch(1)
+        val obsoleteStage = TestExchangeQueryStageExec(
+          105,
+          exchange,
+          exchange.canonicalized,
+          cancelCallback = Some(() => {
+            cancellationEntered.countDown()
+            assert(releaseCancellation.await(30, TimeUnit.SECONDS),
+              "failed stage cancellation was not released")
+            throw new IllegalStateException("test concurrent stage 
cancellation failed")
+          }))
+        adaptivePlan.context.stageCache.put(exchange.canonicalized, 
obsoleteStage)
+
+        val cancelledIds = new AtomicReference[Seq[Int]]()
+        val acquiredStage = new AtomicReference[ExchangeQueryStageExec]()
+        val cancellationWorker = startWorker("aqe-failed-stage-cancellation") {
+          cancelledIds.set(adaptivePlan.invokePrivate(
+            cancelObsoleteStages(replacementPlan, Seq(obsoleteStage))))
+        }
+        var reuseWorker: Option[LifecycleWorker] = None
+
+        try {
+          assert(cancellationEntered.await(30, TimeUnit.SECONDS),
+            "failing stage cancellation did not start")
+          reuseWorker = 
Some(startWorker("aqe-stage-reuse-after-failed-cancellation") {
+            acquiredStage.set(acquireStage(adaptivePlan, exchange))
+          })
+          eventually(timeout(10.seconds), interval(10.milliseconds)) {
+            assert(Set(Thread.State.BLOCKED, Thread.State.WAITING, 
Thread.State.TIMED_WAITING)
+              .contains(reuseWorker.get._1.getState))
+            assert(acquiredStage.get() == null)
+          }
+        } finally {
+          releaseCancellation.countDown()
+          waitForWorker(cancellationWorker)
+          reuseWorker.foreach(waitForWorker)
+        }
+
+        checkWorker(cancellationWorker)
+        reuseWorker.foreach(checkWorker)
+        assert(cancelledIds.get().isEmpty)
+        assert(!obsoleteStage.cancelled)
+        assert(acquiredStage.get().resultOption.eq(obsoleteStage.resultOption))
+        
assert(adaptivePlan.context.isSharedStageResult(obsoleteStage.resultOption))
+        assert(adaptivePlan.context.stageCache.get(exchange.canonicalized)
+          .exists(_ eq obsoleteStage))
+      }
+
+      {
+        val (adaptivePlan, exchange) = newPlanAndExchange()
+        val obsoleteStage = TestExchangeQueryStageExec(
+          102,
+          exchange,
+          exchange.canonicalized,
+          cancelFailure = Some(new IllegalStateException("test stage 
cancellation failed")))
+        adaptivePlan.context.stageCache.put(exchange.canonicalized, 
obsoleteStage)
+
+        val cancelledIds = adaptivePlan.invokePrivate(
+          cancelObsoleteStages(replacementPlan, Seq(obsoleteStage)))
+        assert(cancelledIds.isEmpty)
+        assert(!obsoleteStage.cancelled)
+        assert(adaptivePlan.context.stageCache.get(exchange.canonicalized)
+          .exists(_ eq obsoleteStage))
+
+        val reusedStage = acquireStage(adaptivePlan, exchange)
+        assert(reusedStage.resultOption.eq(obsoleteStage.resultOption))
+        
assert(adaptivePlan.context.isSharedStageResult(obsoleteStage.resultOption))
+
+        obsoleteStage.error.set(Some(new IllegalStateException("shared stage 
failed")))
+        
assert(!adaptivePlan.invokePrivate(ignoreFailedObsoleteStageFailure(obsoleteStage)))
+        assert(adaptivePlan.context.stageCache.get(exchange.canonicalized)
+          .exists(_ eq obsoleteStage))
+      }
+
+      {
+        val (adaptivePlan, exchange) = newPlanAndExchange()
+        val obsoleteStage = TestExchangeQueryStageExec(
+          103,
+          exchange,
+          exchange.canonicalized,
+          cancelFailure = Some(new IllegalStateException("test stage 
cancellation failed")))
+        adaptivePlan.context.stageCache.put(exchange.canonicalized, 
obsoleteStage)
+
+        val cancelledIds = adaptivePlan.invokePrivate(
+          cancelObsoleteStages(replacementPlan, Seq(obsoleteStage)))
+        assert(cancelledIds.isEmpty)
+        assert(adaptivePlan.context.stageCache.get(exchange.canonicalized)
+          .exists(_ eq obsoleteStage))
+
+        obsoleteStage.error.set(Some(new IllegalStateException("obsolete stage 
failed")))
+        
assert(adaptivePlan.invokePrivate(ignoreFailedObsoleteStageFailure(obsoleteStage)))
+        
assert(!adaptivePlan.context.stageCache.contains(exchange.canonicalized))
+      }
+    }
+  }
+
   test("Scalar subquery") {
     withSQLConf(
         SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
@@ -4304,6 +5362,41 @@ class AdaptiveQueryExecSuite
   }
 }
 
+private case class TestExchangeQueryStageExec(
+    override val id: Int,
+    override val plan: SparkPlan,
+    override val _canonicalized: SparkPlan,
+    runtimeRowCount: Option[BigInt] = None,
+    cancelFailure: Option[Throwable] = None,
+    reuseCallback: Option[() => Unit] = None,
+    cancelCallback: Option[() => Unit] = None) extends ExchangeQueryStageExec {
+  @volatile var cancelled: Boolean = false
+
+  override protected def doMaterialize(): Future[Any] = Future.successful(())
+
+  override def getRuntimeStatistics: 
org.apache.spark.sql.catalyst.plans.logical.Statistics =
+    org.apache.spark.sql.catalyst.plans.logical.Statistics(
+      sizeInBytes = BigInt(0), rowCount = runtimeRowCount)
+
+  override protected def doCancel(reason: String): Unit = {
+    cancelFailure.foreach { failure =>
+      throw failure
+    }
+    cancelCallback.foreach(_())
+    cancelled = true
+  }
+
+  override def newReuseInstance(
+      newStageId: Int,
+      newOutput: Seq[Attribute]): ExchangeQueryStageExec = {
+    reuseCallback.foreach(_())
+    val reuse = copy(id = newStageId)
+    reuse._resultOption = this._resultOption
+    reuse._error = this._error
+    reuse
+  }
+}
+
 /**
  * A minimal leaf plan with a single output attribute, used to build tiny 
plans for cost tests.
  */


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org