# UNDF: UNDF-2026-000000535
# spark-0004 — Spark DAGScheduler: waitingStages.filter(_.parents.contains(parent)) O(W×P) on every stage completion

## Metadata
- **Project**: Apache Spark
- **Component**: `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala`
- **CWE**: CWE-407 (Inefficient Algorithmic Complexity)
- **Severity**: MEDIUM
- **Complexity**: O(W × P) per stage completion → O(S × W × P) total where S = stages completing, W = waiting stages, P = parents per stage
- **Hot path**: `submitWaitingChildStages()` is called every time a stage finishes

## Location

```
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
method: submitWaitingChildStages()  line ~1230
```

### Defective code

```scala
// waitingStages is HashSet[Stage] — iteration is O(W)
// Stage.parents is List[Stage] — contains() is O(P) linear scan
val childStages = waitingStages.filter(_.parents.contains(parent)).toArray
```

`Stage.parents` is declared as `List[Stage]` (Scala immutable linked list). The `List.contains()` method performs a linear scan, comparing each element with the target stage by reference equality. The outer `filter` iterates all W waiting stages. For each waiting stage with P parents:
- `parents.contains(parent)` costs O(P)
- Total per completion event: O(W × P)

For a complex Spark job with 200 waiting stages and 5 parents each, this fires 1,000 comparisons for every single stage completion. With S stages completing total, the cumulative cost is O(S × W × P).

### Stage class declaration

```scala
// core/src/main/scala/org/apache/spark/scheduler/Stage.scala line 60
val parents: List[Stage],
```

`List[Stage]` is Scala's singly-linked immutable list — `contains` is O(N) linear scan.

## Fix

Two complementary approaches:

**Option A — Build a reverse adjacency map at job submission time:**

```scala
// In DAGScheduler, maintain a child → Set[Stage] map updated when stages are created:
private val stageChildMap = new HashMap[Stage, HashSet[Stage]]()

// In createShuffleMapStage / createResultStage, populate:
stage.parents.foreach { parent =>
  stageChildMap.getOrElseUpdate(parent, new HashSet[Stage]()) += stage
}

// In submitWaitingChildStages:
val directChildren = stageChildMap.getOrElse(parent, Set.empty)
val childStages = waitingStages.intersect(directChildren).toArray
```

O(1) lookup in the child map, O(intersection) = O(min(W, children)).

**Option B — Use a Set for Stage.parents:**

```scala
// Change Stage.parents from List[Stage] to Set[Stage]:
val parents: Set[Stage],
// Then parents.contains(parent) is O(1) hash lookup
```

Option A is preferred — it eliminates the outer iteration over all waiting stages entirely.

## Complexity analysis

| Scenario | Before | After (Option A) |
|----------|--------|-----------------|
| W waiting, P parents, S completing | O(S × W × P) | O(S × children_count) |
| W=200, P=5, S=200 | 200,000 ops | ~200 ops (avg 1 child) |
| W=500, P=10, S=500 | 2,500,000 ops | ~500 ops |
| Speedup at W=200, P=5 | — | ~1000× |

## Notes

This defect is triggered in any Spark job with a wide DAG — many stages waiting simultaneously, each with multiple parent stages. Data pipeline jobs with complex multi-hop transformations and join-heavy queries are the primary affected workload. The linear scan through `parents` for each of the waiting stages is unnecessary because a reverse adjacency index can be maintained incrementally at zero extra cost during stage creation.
