shrirangmhalgi commented on code in PR #55839:
URL: https://github.com/apache/spark/pull/55839#discussion_r3405045793
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -395,6 +401,39 @@ case class AdaptiveSparkPlanExec(
.get.asInstanceOf[T]
}
+ private def cancelObsoleteStages(
+ newPhysicalPlan: SparkPlan,
+ stagesToReplace: Seq[QueryStageExec]): Seq[Int] = {
+ val newStages = newPhysicalPlan.collect {
+ case stage: QueryStageExec => stage
+ }
+ val obsoleteStages = stagesToReplace.collect {
+ case stage: ExchangeQueryStageExec
+ if !newStages.exists(newStage =>
+ newStage.id == stage.id ||
newStage.resultOption.eq(stage.resultOption)) => stage
+ }
+ obsoleteStages.foreach { stage =>
+ if (!stage.isMaterialized &&
!context.isSharedStageResult(stage.resultOption)) {
+ removeStageFromCache(stage)
+ try {
+ stage.cancel("The query stage is no longer referenced by the current
adaptive plan.")
+ } catch {
+ case NonFatal(t) =>
+ logError(s"Exception in cancelling obsolete query stage:
${stage.treeString}", t)
+ }
+ }
+ }
+ obsoleteStages.map(_.id)
Review Comment:
`cancelObsoleteStages` returns IDs of all obsolete stages, but only a subset
is actually cancelled (those passing `!stage.isMaterialized &&
!context.isSharedStageResult(...))`. These IDs feed into
`obsoleteCancelledStageIds` which suppresses `StageFailure` errors. If an
obsolete-but-not-cancelled stage later reports a failure, the error would be
silently swallowed. Should the returned IDs be limited to stages that actually
entered the `cancel()` branch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
