java-topology/defects/kubeflow/patch/kubeflow-0001-pipeline-spec-builder-tasks-list.md

3.9 KiB
Raw Blame History

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 13831540 Outer subgroup loop rebuilding list
sdk/python/kfp/compiler/pipeline_spec_builder.py 91300 build_task_spec_for_task — multiple O(T) in checks
sdk/python/kfp/compiler/pipeline_spec_builder.py 11921260 build_task_spec_for_group — O(T) in checks

Defective Code

pipeline_spec_builder.py lines 13831395 (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

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