# UNDF: UNDF-2026-000000442
# kubeflow-0001 — O(T²×I) Pipeline Compiler: tasks_in_current_dag List membership in DAG compilation

**Severity:** HIGH
**Complexity:** O(T²×I) → O(T×I)
**CWE:** CWE-407 (Algorithmic Complexity)

## Affected File

| File | Lines | Notes |
|------|-------|-------|
| `sdk/python/kfp/compiler/pipeline_spec_builder.py` | 1383–1540 | Outer subgroup loop rebuilding list |
| `sdk/python/kfp/compiler/pipeline_spec_builder.py` | 91–300 | `build_task_spec_for_task` — multiple O(T) `in` checks |
| `sdk/python/kfp/compiler/pipeline_spec_builder.py` | 1192–1260 | `build_task_spec_for_group` — O(T) `in` checks |

## Defective Code

### `pipeline_spec_builder.py` lines 1383–1395 (outer compilation loop)

```python
subgroups = group.groups + group.tasks
for subgroup in subgroups:                         # O(T) outer loop
    ...
    tasks_in_current_dag = [                       # ← REBUILT every iteration: O(T)
        utils.sanitize_task_name(subgroup.name) for subgroup in subgroups
    ]
    ...
    if isinstance(subgroup, pipeline_task.PipelineTask):
        subgroup_task_spec = build_task_spec_for_task(
            task=subgroup,
            parent_component_inputs=...,
            tasks_in_current_dag=tasks_in_current_dag,   # List[str] passed in
        )
```

### `pipeline_spec_builder.py` lines 194, 229, 293 (inside build_task_spec_for_task)

```python
for input_name, input_value in task.inputs.items():     # O(I) inputs per task
    ...
    if input_value.task_name in tasks_in_current_dag:   # ← O(T) scan of List[str]
        ...
    if input_value.task_name in tasks_in_current_dag:   # ← O(T) scan again
        ...
    for channel in pipeline_channels:
        if channel.task_name in tasks_in_current_dag:   # ← O(T) scan again
```

**Defect:** `tasks_in_current_dag` is typed as `List[str]` (line 91 signature) and constructed
via list comprehension on every iteration of the outer `for subgroup in subgroups` loop — even
though the set of tasks in the DAG is fixed for the entire loop body.

Inside `build_task_spec_for_task`, each input channel performs an O(T) linear `in` search against
this list. With T tasks, I inputs per task, and 3 separate `in` checks per input branch:

- Total cost: O(T) rebuild × ignored (constant factor) + O(T) outer × O(I) inner × O(T) in-check
- **= O(T² × I)**

## Fix

1. Hoist the `tasks_in_current_dag` computation outside the loop (build once).
2. Change the type from `List[str]` to `Set[str]` for O(1) membership.

```python
# Build once, outside the loop
tasks_in_current_dag: Set[str] = {
    utils.sanitize_task_name(sg.name) for sg in subgroups
}

for subgroup in subgroups:
    ...
    # Pass the pre-built set — no more O(T) rebuild or O(T) in-check
    if isinstance(subgroup, pipeline_task.PipelineTask):
        subgroup_task_spec = build_task_spec_for_task(
            task=subgroup,
            parent_component_inputs=group_component_spec.input_definitions,
            tasks_in_current_dag=tasks_in_current_dag,   # now Set[str]
        )
```

Update the function signatures:

```python
def build_task_spec_for_task(
    task: pipeline_task.PipelineTask,
    parent_component_inputs: pipeline_spec_pb2.ComponentInputsSpec,
    tasks_in_current_dag: Set[str],    # was List[str]
) -> pipeline_spec_pb2.PipelineTaskSpec:
```

The `in` checks at lines 194, 229, 293, 1235, 1247 now run in O(1) instead of O(T).

## Complexity

| Step | Before | After |
|------|--------|-------|
| Build `tasks_in_current_dag` | O(T) × T iterations = O(T²) | O(T) once |
| Each `in` check | O(T) | O(1) |
| Total compile cost per DAG | O(T² × I) | O(T × I) |

## Impact

Kubeflow Pipelines pipelines with hundreds of components (common in large ML training workflows,
feature engineering pipelines, and AutoML grids) pay quadratic cost during SDK compile time.
A pipeline with 200 tasks and 5 inputs each: 200² × 5 × 3 checks = 600,000 operations vs 3,000
with the fix — a 200× reduction.
