java-topology/defects/spark/patch/spark-0002-dagscheduler-listbuffer-remove0-arraydeque.md

# 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.