# UNDF: UNDF-2026-000000534 # spark-0002 — DAGScheduler BFS queues: ListBuffer.remove(0) is O(N) → O(N²) total **Severity:** HIGH **CWE:** CWE-407 (Algorithmic Complexity — Inefficient Algorithmic Complexity) **File:** `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` **Introduced:** Long-standing; present in all recent Spark versions ## Defect Six BFS-style graph traversal functions in `DAGScheduler` use `scala.collection.mutable.ListBuffer` as the work queue and dequeue with `waitingForVisit.remove(0)`. `ListBuffer.remove(0)` is **O(N)** — it shifts every remaining element left by one position. Called inside a `while (waitingForVisit.nonEmpty)` loop over N RDD nodes, this makes the overall traversal **O(N²)** instead of O(N). For a DAG with hundreds of RDD partitions or deep lineage chains (e.g. iterative ML workloads, complex SQL plans), this is the dominant scheduling cost. ### Affected functions and exact lines (Spark master, 2026-03) | Function | `waitingForVisit` init | `remove(0)` line | |---|---|---| | `getMissingAncestorShuffleDependencies` | 694 | 696 | | `getShuffleDependenciesAndResourceProfiles` | 730 | 732 | | `traverseParentRDDsWithinStage` | 754 | 756 | | `getMissingParentStages` | 779 | 816 | | `eagerlyComputePartitionsForRddAndAncestors` | 828 | 844 | | `stageDependsOn` | 3381 | 3399 | ### Code pattern (repeated 6 times) ```scala // BEFORE — O(N²) BFS val waitingForVisit = new ListBuffer[RDD[_]] waitingForVisit += rdd while (waitingForVisit.nonEmpty) { val toVisit = waitingForVisit.remove(0) // ← O(N) shift every iteration ... waitingForVisit.prepend(dependency.rdd) // O(1) prepend, but dequeue dominates } ``` ## Fix Replace `ListBuffer` with `scala.collection.mutable.ArrayDeque`, which provides **O(1) amortized** prepend (`prepend`) and dequeue (`removeHead()`). ```scala // AFTER — O(N) BFS val waitingForVisit = new mutable.ArrayDeque[RDD[_]]() waitingForVisit += rdd while (waitingForVisit.nonEmpty) { val toVisit = waitingForVisit.removeHead() // ← O(1) ... waitingForVisit.prepend(dependency.rdd) // ← O(1) } ``` `ArrayDeque` was added to the Scala standard library in 2.13. Spark already targets Scala 2.13+, so no new dependency is introduced. The `ListBuffer` import can be removed from `DAGScheduler.scala` once all six sites are migrated (it is not used for any other purpose in the file). ## Patch ```diff --- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala @@ -29,7 +29,7 @@ import scala.collection.mutable -import scala.collection.mutable.{HashMap, HashSet, ListBuffer} +import scala.collection.mutable.{ArrayDeque, HashMap, HashSet} // getMissingAncestorShuffleDependencies (~line 688) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += rdd while (waitingForVisit.nonEmpty) { - val toVisit = waitingForVisit.remove(0) + val toVisit = waitingForVisit.removeHead() // getShuffleDependenciesAndResourceProfiles (~line 724) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += rdd while (waitingForVisit.nonEmpty) { - val toVisit = waitingForVisit.remove(0) + val toVisit = waitingForVisit.removeHead() // traverseParentRDDsWithinStage (~line 749) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += rdd while (waitingForVisit.nonEmpty) { - val toVisit = waitingForVisit.remove(0) + val toVisit = waitingForVisit.removeHead() // getMissingParentStages (~line 779) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += stage.rdd // ... (visit function defined inline) while (waitingForVisit.nonEmpty) { - visit(waitingForVisit.remove(0)) + visit(waitingForVisit.removeHead()) // eagerlyComputePartitionsForRddAndAncestors (~line 828) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += rdd // ... (visit function defined inline) while (waitingForVisit.nonEmpty) { - visit(waitingForVisit.remove(0)) + visit(waitingForVisit.removeHead()) // stageDependsOn (~line 3377) - val waitingForVisit = new ListBuffer[RDD[_]] + val waitingForVisit = new ArrayDeque[RDD[_]]() waitingForVisit += stage.rdd // ... (visit function defined inline) while (waitingForVisit.nonEmpty) { - visit(waitingForVisit.remove(0)) + visit(waitingForVisit.removeHead()) ``` ## Complexity | Metric | Before | After | |---|---|---| | BFS over N RDD nodes | O(N²) | O(N) | | `remove(0)` / `removeHead()` | O(N) per call | O(1) amortized | | Memory | O(N) | O(N) | ## Impact - Affects every Spark job submission: `getMissingParentStages` is called on every `submitStage` - `stageDependsOn` called on stage resubmission after failure — worst case is already degraded - Iterative ML pipelines (Spark MLlib) with deep RDD lineage chains: `eagerlyComputePartitionsForRddAndAncestors` called per action - Complex SQL joins: `getShuffleDependenciesAndResourceProfiles` called per query plan ## Test `defects/spark/unit/SparkDAGSchedulerTest.java` — runs slow (ListBuffer simulation) vs fast (ArrayDeque simulation), verifies O(N²) vs O(N) operation counts.