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)

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)

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

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

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

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.

3.9 KiB Raw Blame History Unescape Escape