You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "jiangxb1987 (via GitHub)" <gi...@apache.org> on 2026/04/15 23:30:14 UTC
[PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
jiangxb1987 opened a new pull request, #55361:
URL: https://github.com/apache/spark/pull/55361
### What changes were proposed in this pull request?
- Introduced two private helper methods in `DAGScheduler`:
- `traverseRDDGraph(rdd)(visitor)` — traverses the RDD dependency graph using an explicit stack, calling `visitor(rdd, enqueue)` for each unvisited RDD.
- `traverseRDDGraphUntil(rdd)(visitor)` — like above but supports early termination: the visitor returns `false` to stop traversal; returns whether traversal completed.
- Refactored 6 methods to use these helpers, eliminating duplicated `visited + waitingForVisit` boilerplate:
- `getMissingAncestorShuffleDependencies`
- `getShuffleDependenciesAndResourceProfiles`
- `traverseParentRDDsWithinStage`
- `getMissingParentStages`
- `eagerlyComputePartitionsForRddAndAncestors`
- `stageDependsOn`
### Why are the changes needed?
The same BFS traversal pattern (maintain a `visited` set and a `waitingForVisit` stack, remove-and-check-visited in a loop) was duplicated across many methods in `DAGScheduler`, adding unnecessary boilerplate and making the code harder to read and maintain.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Covered by existing `DAGSchedulerSuite` tests. No behavior changes were made.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude (claude-sonnet-4-6)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3090604241
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -683,28 +683,55 @@ private[spark] class DAGScheduler(
}.toList
}
- /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
- private def getMissingAncestorShuffleDependencies(
- rdd: RDD[_]): ListBuffer[ShuffleDependency[_, _, _]] = {
- val ancestors = new ListBuffer[ShuffleDependency[_, _, _]]
+ /**
+ * Traverses the RDD dependency graph using a manually maintained stack to prevent
+ * StackOverflowError caused by recursive traversal. For each unvisited RDD, calls
Review Comment:
Maybe we should move the comment "Traverses the RDD dependency graph using a manually maintained stack to prevent StackOverflowError caused by recursive traversal." to `traverseRDDGraphUntil()`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3096374316
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -749,100 +770,68 @@ private[spark] class DAGScheduler(
* RDDs satisfy a given predicate.
*/
private def traverseParentRDDsWithinStage(rdd: RDD[_], predicate: RDD[_] => Boolean): Boolean = {
- val visited = new HashSet[RDD[_]]
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += rdd
- while (waitingForVisit.nonEmpty) {
- val toVisit = waitingForVisit.remove(0)
- if (!visited(toVisit)) {
- if (!predicate(toVisit)) {
- return false
- }
- visited += toVisit
+ traverseRDDGraphUntil(rdd) { (toVisit, enqueue) =>
+ if (!predicate(toVisit)) {
+ false
+ } else {
toVisit.dependencies.foreach {
case _: ShuffleDependency[_, _, _] =>
- // Not within the same stage with current rdd, do nothing.
+ // Not within the same stage as the current RDD, do nothing.
case dependency =>
- waitingForVisit.prepend(dependency.rdd)
+ enqueue(dependency.rdd)
}
+ true
}
}
- true
}
private def getMissingParentStages(stage: Stage): List[Stage] = {
val missing = new HashSet[Stage]
- val visited = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += stage.rdd
- def visit(rdd: RDD[_]): Unit = {
- if (!visited(rdd)) {
- visited += rdd
- val rddHasUncachedPartitions = try {
- getCacheLocs(rdd).contains(Nil)
- } catch {
- case e: RpcTimeoutException =>
- logWarning(log"Failed to get cache locations for RDD ${MDC(RDD_ID, rdd.id)} due " +
- log"to rpc timeout, assuming not fully cached.", e)
- true
- }
- if (rddHasUncachedPartitions) {
- for (dep <- rdd.dependencies) {
- dep match {
- case shufDep: ShuffleDependency[_, _, _] =>
- val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
- // Mark mapStage as available with shuffle outputs only after shuffle merge is
- // finalized with push based shuffle. If not, subsequent ShuffleMapStage won't
- // read from merged output as the MergeStatuses are not available.
- if (!mapStage.isAvailable || !mapStage.shuffleDep.shuffleMergeFinalized) {
- missing += mapStage
- } else {
- // Forward the nextAttemptId if skipped and get visited for the first time.
- // Otherwise, once it gets retried,
- // 1) the stuffs in stage info become distorting, e.g. task num, input byte, e.t.c
- // 2) the first attempt starts from 0-idx, it will not be marked as a retry
- mapStage.increaseAttemptIdOnFirstSkip()
- }
- case narrowDep: NarrowDependency[_] =>
- waitingForVisit.prepend(narrowDep.rdd)
- }
+ traverseRDDGraph(stage.rdd) { (rdd, enqueue) =>
+ val rddHasUncachedPartitions = try {
+ getCacheLocs(rdd).contains(Nil)
+ } catch {
+ case e: RpcTimeoutException =>
+ logWarning(log"Failed to get cache locations for RDD ${MDC(RDD_ID, rdd.id)} due " +
+ log"to rpc timeout, assuming not fully cached.", e)
+ true
+ }
+ if (rddHasUncachedPartitions) {
+ for (dep <- rdd.dependencies) {
Review Comment:
This matches the original behavior, I don't think we should change it in this PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3097868803
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -683,28 +683,56 @@ private[spark] class DAGScheduler(
}.toList
}
- /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
- private def getMissingAncestorShuffleDependencies(
- rdd: RDD[_]): ListBuffer[ShuffleDependency[_, _, _]] = {
- val ancestors = new ListBuffer[ShuffleDependency[_, _, _]]
+ /**
+ * Like [[traverseRDDGraphUntil]], but does not support early termination. For each unvisited RDD,
+ * calls `visitor(rdd, enqueue)` where `enqueue` can be called to schedule additional RDDs for
+ * traversal.
+ */
+ private def traverseRDDGraph(rdd: RDD[_])(visitor: (RDD[_], RDD[_] => Unit) => Unit): Unit = {
+ traverseRDDGraphUntil(rdd) { (r, enqueue) =>
+ visitor(r, enqueue)
+ true
+ }
+ }
+
+ /**
+ * Traverses the RDD dependency graph using a manually maintained stack to prevent
+ * StackOverflowError caused by recursive traversal. For each unvisited RDD, calls
+ * `visitor(rdd, enqueue)` where `enqueue` can be called to schedule additional RDDs for
+ * traversal. If `visitor` returns `false`, the traversal stops immediately. Returns `true`
+ * if the traversal completed normally, `false` if it was terminated early by the visitor.
+ */
+ private def traverseRDDGraphUntil(
+ rdd: RDD[_])(visitor: (RDD[_], RDD[_] => Unit) => Boolean): Boolean = {
val visited = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
val waitingForVisit = new ListBuffer[RDD[_]]
waitingForVisit += rdd
+ def enqueue(r: RDD[_]): Unit = waitingForVisit.prepend(r)
while (waitingForVisit.nonEmpty) {
val toVisit = waitingForVisit.remove(0)
if (!visited(toVisit)) {
visited += toVisit
- val (shuffleDeps, _) = getShuffleDependenciesAndResourceProfiles(toVisit)
- shuffleDeps.foreach { shuffleDep =>
- if (!shuffleIdToMapStage.contains(shuffleDep.shuffleId)) {
- ancestors.prepend(shuffleDep)
- waitingForVisit.prepend(shuffleDep.rdd)
- } // Otherwise, the dependency and its ancestors have already been registered.
+ if (!visitor(toVisit, enqueue)) {
+ return false
}
}
}
+ true
+ }
+
+ /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
Review Comment:
The field is `shuffleIdToMapStage` (line 160). Since this comment is re-added in the diff, worth fixing the stale name while we're here.
```suggestion
/** Find ancestor shuffle dependencies that are not registered in shuffleIdToMapStage yet */
```
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -3374,31 +3363,27 @@ private[spark] class DAGScheduler(
if (stage == target) {
return true
}
- val visitedRdds = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += stage.rdd
- def visit(rdd: RDD[_]): Unit = {
- if (!visitedRdds(rdd)) {
- visitedRdds += rdd
+ var found = false
Review Comment:
Minor: `traverseRDDGraphUntil` already returns `false` iff the visitor terminated early, so we can drop the `var found` and return the negation of the helper directly — matching the shape of `traverseParentRDDsWithinStage` which returns the helper's result directly:
```scala
!traverseRDDGraphUntil(stage.rdd) { (rdd, enqueue) =>
if (rdd == target.rdd) {
false
} else {
for (dep <- rdd.dependencies) {
dep match {
case shufDep: ShuffleDependency[_, _, _] =>
val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
if (!mapStage.isAvailable) {
enqueue(mapStage.rdd)
} // Otherwise there's no need to follow the dependency back
case narrowDep: NarrowDependency[_] =>
enqueue(narrowDep.rdd)
}
}
true
}
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on PR #55361:
URL: https://github.com/apache/spark/pull/55361#issuecomment-4284147744
Thanks, merging to master!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #55361: [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler
URL: https://github.com/apache/spark/pull/55361
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3103273494
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -683,28 +683,56 @@ private[spark] class DAGScheduler(
}.toList
}
- /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
- private def getMissingAncestorShuffleDependencies(
- rdd: RDD[_]): ListBuffer[ShuffleDependency[_, _, _]] = {
- val ancestors = new ListBuffer[ShuffleDependency[_, _, _]]
+ /**
+ * Like [[traverseRDDGraphUntil]], but does not support early termination. For each unvisited RDD,
+ * calls `visitor(rdd, enqueue)` where `enqueue` can be called to schedule additional RDDs for
+ * traversal.
+ */
+ private def traverseRDDGraph(rdd: RDD[_])(visitor: (RDD[_], RDD[_] => Unit) => Unit): Unit = {
+ traverseRDDGraphUntil(rdd) { (r, enqueue) =>
+ visitor(r, enqueue)
+ true
+ }
+ }
+
+ /**
+ * Traverses the RDD dependency graph using a manually maintained stack to prevent
+ * StackOverflowError caused by recursive traversal. For each unvisited RDD, calls
+ * `visitor(rdd, enqueue)` where `enqueue` can be called to schedule additional RDDs for
+ * traversal. If `visitor` returns `false`, the traversal stops immediately. Returns `true`
+ * if the traversal completed normally, `false` if it was terminated early by the visitor.
+ */
+ private def traverseRDDGraphUntil(
+ rdd: RDD[_])(visitor: (RDD[_], RDD[_] => Unit) => Boolean): Boolean = {
val visited = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
val waitingForVisit = new ListBuffer[RDD[_]]
waitingForVisit += rdd
+ def enqueue(r: RDD[_]): Unit = waitingForVisit.prepend(r)
while (waitingForVisit.nonEmpty) {
val toVisit = waitingForVisit.remove(0)
if (!visited(toVisit)) {
visited += toVisit
- val (shuffleDeps, _) = getShuffleDependenciesAndResourceProfiles(toVisit)
- shuffleDeps.foreach { shuffleDep =>
- if (!shuffleIdToMapStage.contains(shuffleDep.shuffleId)) {
- ancestors.prepend(shuffleDep)
- waitingForVisit.prepend(shuffleDep.rdd)
- } // Otherwise, the dependency and its ancestors have already been registered.
+ if (!visitor(toVisit, enqueue)) {
+ return false
}
}
}
+ true
+ }
+
+ /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
Review Comment:
Fixed, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3103275008
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -3374,31 +3363,27 @@ private[spark] class DAGScheduler(
if (stage == target) {
return true
}
- val visitedRdds = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += stage.rdd
- def visit(rdd: RDD[_]): Unit = {
- if (!visitedRdds(rdd)) {
- visitedRdds += rdd
+ var found = false
Review Comment:
Good call, updated to return `!traverseRDDGraphUntil(...)` directly.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "lakechd (via GitHub)" <gi...@apache.org>.
lakechd commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3096203307
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -749,100 +770,68 @@ private[spark] class DAGScheduler(
* RDDs satisfy a given predicate.
*/
private def traverseParentRDDsWithinStage(rdd: RDD[_], predicate: RDD[_] => Boolean): Boolean = {
- val visited = new HashSet[RDD[_]]
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += rdd
- while (waitingForVisit.nonEmpty) {
- val toVisit = waitingForVisit.remove(0)
- if (!visited(toVisit)) {
- if (!predicate(toVisit)) {
- return false
- }
- visited += toVisit
+ traverseRDDGraphUntil(rdd) { (toVisit, enqueue) =>
+ if (!predicate(toVisit)) {
+ false
+ } else {
toVisit.dependencies.foreach {
case _: ShuffleDependency[_, _, _] =>
- // Not within the same stage with current rdd, do nothing.
+ // Not within the same stage as the current RDD, do nothing.
case dependency =>
- waitingForVisit.prepend(dependency.rdd)
+ enqueue(dependency.rdd)
}
+ true
}
}
- true
}
private def getMissingParentStages(stage: Stage): List[Stage] = {
val missing = new HashSet[Stage]
- val visited = new HashSet[RDD[_]]
- // We are manually maintaining a stack here to prevent StackOverflowError
- // caused by recursively visiting
- val waitingForVisit = new ListBuffer[RDD[_]]
- waitingForVisit += stage.rdd
- def visit(rdd: RDD[_]): Unit = {
- if (!visited(rdd)) {
- visited += rdd
- val rddHasUncachedPartitions = try {
- getCacheLocs(rdd).contains(Nil)
- } catch {
- case e: RpcTimeoutException =>
- logWarning(log"Failed to get cache locations for RDD ${MDC(RDD_ID, rdd.id)} due " +
- log"to rpc timeout, assuming not fully cached.", e)
- true
- }
- if (rddHasUncachedPartitions) {
- for (dep <- rdd.dependencies) {
- dep match {
- case shufDep: ShuffleDependency[_, _, _] =>
- val mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
- // Mark mapStage as available with shuffle outputs only after shuffle merge is
- // finalized with push based shuffle. If not, subsequent ShuffleMapStage won't
- // read from merged output as the MergeStatuses are not available.
- if (!mapStage.isAvailable || !mapStage.shuffleDep.shuffleMergeFinalized) {
- missing += mapStage
- } else {
- // Forward the nextAttemptId if skipped and get visited for the first time.
- // Otherwise, once it gets retried,
- // 1) the stuffs in stage info become distorting, e.g. task num, input byte, e.t.c
- // 2) the first attempt starts from 0-idx, it will not be marked as a retry
- mapStage.increaseAttemptIdOnFirstSkip()
- }
- case narrowDep: NarrowDependency[_] =>
- waitingForVisit.prepend(narrowDep.rdd)
- }
+ traverseRDDGraph(stage.rdd) { (rdd, enqueue) =>
+ val rddHasUncachedPartitions = try {
+ getCacheLocs(rdd).contains(Nil)
+ } catch {
+ case e: RpcTimeoutException =>
+ logWarning(log"Failed to get cache locations for RDD ${MDC(RDD_ID, rdd.id)} due " +
+ log"to rpc timeout, assuming not fully cached.", e)
+ true
+ }
+ if (rddHasUncachedPartitions) {
+ for (dep <- rdd.dependencies) {
Review Comment:
it will loop over all the dependencies, regardless the path has been visited before
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56499][CORE] Deduplicate RDD graph BFS traversal pattern in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55361:
URL: https://github.com/apache/spark/pull/55361#discussion_r3095467483
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -683,28 +683,55 @@ private[spark] class DAGScheduler(
}.toList
}
- /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
- private def getMissingAncestorShuffleDependencies(
- rdd: RDD[_]): ListBuffer[ShuffleDependency[_, _, _]] = {
- val ancestors = new ListBuffer[ShuffleDependency[_, _, _]]
+ /**
+ * Traverses the RDD dependency graph using a manually maintained stack to prevent
+ * StackOverflowError caused by recursive traversal. For each unvisited RDD, calls
Review Comment:
updated
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org