otterc commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r640073914
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -2136,9 +2137,24 @@ private[spark] class DAGScheduler(
}
}
- private[scheduler] def handleShuffleMergeFinalized(stage: ShuffleMapStage):
Unit = {
- stage.shuffleDep.markShuffleMergeFinalized
- processShuffleMapStageCompletion(stage)
+ private[scheduler] def handleRegisterMergeStatuses(
+ stage: ShuffleMapStage,
+ mergeStatuses: Seq[(Int, MergeStatus)]): Unit = {
+ // Register merge statuses if the stage is still running and shuffle merge
is not finalized yet.
+ if (runningStages.contains(stage) &&
!stage.shuffleDep.shuffleMergeFinalized) {
+ mapOutputTracker.registerMergeResults(stage.shuffleDep.shuffleId,
mergeStatuses)
+ }
+ }
+
+ private[scheduler] def handleShuffleMergeFinalized(
+ stage: ShuffleMapStage): Unit = {
+ // Only update MapOutputTracker metadata if the stage is still active. i.e
not cancelled.
+ if (runningStages.contains(stage)) {
+ stage.shuffleDep.markShuffleMergeFinalized()
+ processShuffleMapStageCompletion(stage)
+ } else {
+ mapOutputTracker.unregisterAllMergeResult(stage.shuffleDep.shuffleId)
Review comment:
> Stage will be active until the shuffle merge is finalized and then
only we are processing the map stage completion, isn't it? So if the stage is
not part of running stages and we still reach the handling shuffle merge
finalize, then we need to unregister the merge results, isn't it?
Are you just making an assumption here, that if the stage is not running and
then this finalized message is processed that means the stage is cancelled? Is
this a valid assumption?
If yes, then can you add a comment here.
> Can you think of a scenario where stage is not part of running stages and
still shuffle merge is finalized? - Ideally this should not happen.
This is what is throwing me off. If this is not ideally going to happen then
why are we unregistering the results here. Again, if the assumption is that
this happens when the stage was cancelled then document it. Also, if handling
stage cancellation wrt merge finalization is not handled in this PR then why
have this unregistration of merge results here?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
