HyukjinKwon commented on code in PR #59083:
URL: https://github.com/apache/spark/pull/59083#discussion_r4172687347
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1043,21 +1116,93 @@ private[spark] class MapOutputTrackerMaster(
/**
* Removes all shuffle outputs associated with this host. Note that this
will also remove
- * outputs which are served by an external shuffle server (if one exists).
+ * outputs which are served by an external shuffle server (if one exists).
Unconditional cleanup:
+ * every shuffle's output on the host is dropped. See the two-argument
overload for the selective
+ * (reliable-storage-preserving) variant.
*/
- def removeOutputsOnHost(host: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
- incrementEpoch()
- }
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, respectReliablyStored = false)
/**
- * Removes all shuffle outputs associated with this executor. Note that this
will also remove
+ * Removes all map outputs associated with this executor. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists),
as they are still
- * registered with this execId.
+ * registered with this execId. Unconditional cleanup: every shuffle's
output on the executor is
+ * dropped. See the three-argument overload for the selective variant.
*/
- def removeOutputsOnExecutor(execId: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnExecutor(execId)
}
- incrementEpoch()
+ def removeOutputsOnExecutor(execId: String): Unit =
+ removeOutputsOnExecutor(execId, respectReliablyStored = false)
+
+ /**
+ * Removes shuffle outputs on this host. With `respectReliablyStored` true,
reliably-stored
+ * shuffles are kept. Host-local merge results are always removed.
+ */
+ def removeOutputsOnHost(host: String, respectReliablyStored: Boolean):
CleanupOutcome =
+ removeOutputsOnHost(host, respectReliablyStored, None)
+
+ /**
+ * `failedShuffleId` (the shuffle whose fetch failed) is cleared even when
reliably stored, so its
+ * correlated maps go in one pass. With `restrictToFailedShuffle`, only that
shuffle is touched.
+ */
+ def removeOutputsOnHost(
+ host: String,
+ respectReliablyStored: Boolean,
+ failedShuffleId: Option[Int],
+ restrictToFailedShuffle: Boolean = false): CleanupOutcome = {
+ val outcome = removeSelectively(
+ respectReliablyStored, failedShuffleId, restrictToFailedShuffle)(
+ (status, respect) => status.removeOutputsOnHost(host, respect))
+ if (outcome.shouldBumpEpoch) incrementEpoch()
+ outcome
+ }
+
+ /**
+ * Removes map outputs on this executor. With `respectReliablyStored` true,
reliably-stored
+ * shuffles are kept.
+ */
+ def removeOutputsOnExecutor(execId: String, respectReliablyStored: Boolean):
CleanupOutcome =
+ removeOutputsOnExecutor(execId, respectReliablyStored, None)
+
+ /**
+ * `failedShuffleId` (the shuffle whose fetch failed) is cleared even when
reliably stored, so its
+ * correlated maps go in one pass. With `restrictToFailedShuffle`, only that
shuffle is touched.
+ */
+ def removeOutputsOnExecutor(
+ execId: String,
+ respectReliablyStored: Boolean,
+ failedShuffleId: Option[Int],
+ restrictToFailedShuffle: Boolean = false): CleanupOutcome = {
+ val outcome = removeSelectively(
+ respectReliablyStored, failedShuffleId, restrictToFailedShuffle)(
+ (status, respect) => status.removeOutputsOnExecutor(execId, respect))
+ if (outcome.shouldBumpEpoch) incrementEpoch()
+ outcome
+ }
+
+ /**
+ * Applies `remove` to each shuffle and aggregates the results.
`respectReliablyStored` is passed
+ * per shuffle, except `failedShuffleId` is always cleaned unconditionally.
With
+ * `restrictToFailedShuffle`, only `failedShuffleId` is touched and all
other shuffles are left
+ * intact (used when a reliable-shuffle FetchFailed bypasses the executor
fence: only that shuffle
+ * is proven gone). Does not bump the epoch; see
`CleanupOutcome.shouldBumpEpoch`.
+ */
+ private def removeSelectively(
+ respectReliablyStored: Boolean,
+ failedShuffleId: Option[Int],
+ restrictToFailedShuffle: Boolean = false)(
+ remove: (ShuffleStatus, Boolean) => ShuffleRemovalResult):
CleanupOutcome = {
+ require(!restrictToFailedShuffle || failedShuffleId.isDefined,
+ "restrictToFailedShuffle requires a failedShuffleId to restrict to")
+ var metadataChanged = false
+ var preservedReliable = false
+ shuffleStatuses.foreach { case (shuffleId, status) =>
Review Comment:
**Nit (P3):** In the `restrictToFailedShuffle` path this iterates all of
`shuffleStatuses`, but it only ever acts on the single entry whose `shuffleId
== failedShuffleId.get`. Since the map is keyed by `shuffleId`, a direct
`shuffleStatuses.get(failedShuffleId.get)` is behavior-preserving (ordering is
irrelevant for a single element) and O(1) instead of O(number of active
shuffles) per call. The full iteration is only needed for the unrestricted path.
**Verification:**
- **Regression:** Existing removeSelectively behavior (including the
restricted path) is unchanged under MapOutputTrackerSuite.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1043,21 +1116,93 @@ private[spark] class MapOutputTrackerMaster(
/**
* Removes all shuffle outputs associated with this host. Note that this
will also remove
- * outputs which are served by an external shuffle server (if one exists).
+ * outputs which are served by an external shuffle server (if one exists).
Unconditional cleanup:
+ * every shuffle's output on the host is dropped. See the two-argument
overload for the selective
+ * (reliable-storage-preserving) variant.
*/
- def removeOutputsOnHost(host: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
- incrementEpoch()
- }
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, respectReliablyStored = false)
/**
- * Removes all shuffle outputs associated with this executor. Note that this
will also remove
+ * Removes all map outputs associated with this executor. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists),
as they are still
- * registered with this execId.
+ * registered with this execId. Unconditional cleanup: every shuffle's
output on the executor is
+ * dropped. See the three-argument overload for the selective variant.
Review Comment:
**Nit (P3):** This references a "three-argument overload" for the selective
variant, but there is no three-argument overload -- the selective
(reliable-storage-preserving) variant is the two-argument
`removeOutputsOnExecutor(execId, respectReliablyStored)`. Worth fixing the
reference so it names the overload that exists.
**Verification:**
- **Inspection:** The scaladoc references an overload that exists (the
two-argument selective variant).
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4739,6 +4758,7 @@ private[spark] class DAGScheduler(
executorFailureEpoch -= execId
}
shuffleFileLostEpoch -= execId
+ reliableShuffleFileLostEpoch.filterInPlace { case ((exec, _), _) => exec
!= execId }
Review Comment:
**Nit (P3):** The executor-add retirement of `reliableShuffleFileLostEpoch`
here bounds the map across executor churn, but no test exercises it -- the
existing `reliableShuffleFileLostEpochSize` assertions cover the SPARK-59138
retirement, not this path. Consider a `DAGSchedulerSuite` case that seeds fence
entries for an executor, fires `handleExecutorAdded` for it, and asserts its
entries are retired while unrelated entries remain, so a regression into
unbounded growth has a signal.
**Verification:**
- **Regression:** The test fails if the executor-add retirement stops
retiring the right entries (e.g. unbounded growth).
--
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]
