Clearing a task instance with both "Upstream" and "Downstream" clears unrelated tasks (whole run when the DAG has a shared root)

[ZhiyaD]

Apache Airflow version

3.3.1

What happened?

When clearing a task instance for a specific Dag run with both include_upstream and include_downstream enabled (the "Upstream" + "Downstream" toggles in the Clear Task Instance dialog, or POST /api/v2/dags/{dag_id}/clearTaskInstances with a dag_run_id), Airflow also selects tasks that are neither upstream nor downstream of the selected task.

The downstream traversal starts from the selected task plus all of its upstream tasks, so it picks up every descendant of those ancestors. In a Dag where all branches hang off a shared root task, selecting both options on any task selects every task instance in the run. On a large Dag (~1,300 tasks) this made the dialog propose clearing the entire run instead of one branch.

Selecting only Upstream, or only Downstream, returns the expected set.

What you think should happen instead?

Upstream + Downstream should return the union of the task's own ancestors and the task's own descendants (the Airflow 2 behaviour), not the descendants of its ancestors.

How to reproduce

from airflow.sdk import DAG
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG("clear_repro", schedule=None):
    root = EmptyOperator(task_id="root")
    a = EmptyOperator(task_id="a")
    b = EmptyOperator(task_id="b")
    other = EmptyOperator(task_id="other")
    root >> a >> b
    root >> other

Trigger a run (run1), then dry-run a clear of task a:

POST /api/v2/dags/clear_repro/clearTaskInstances
{"dag_run_id": "run1", "task_ids": ["a"], "include_upstream": ..., "include_downstream": ...,
 "only_failed": false, "dry_run": true}

Result on 3.3.1:

include_upstream	include_downstream	returned task_ids	expected
true	false	a, root	a, root
false	true	a, b	a, b
true	true	a, b, other, root	a, b, root

other is neither upstream nor downstream of a.

Operating System

Debian (official apache/airflow:slim-3.3.1-python3.12 image); reproduced locally on macOS with SQLite.

Versions of Apache Airflow Providers

apache-airflow-providers-standard 1.18.0 (not relevant to the bug)

Deployment

Other Docker-based deployment (Kubernetes, KubernetesExecutor)

Deployment details

No response

Anything else?

It looks like this comes from the dag_run_id branch of post_clear_task_instances (airflow/api_fastapi/core_api/routes/public/task_instances.py):

if upstream:
    _collect_relatives(dag_run_id, "upstream")
if downstream:
    _collect_relatives(dag_run_id, "downstream")

Both calls update normal_task_ids in place, so the downstream pass starts from the set that already contains the upstream tasks. Before #57758 (3.1.4), a single dag.partial_subset(include_upstream=..., include_downstream=...) computed both directions from the original task. The code on main looks the same.

A possible fix: run both directions from a copy of the original seed set and union the results, plus a test that enables both flags together.

Workaround: clear twice, once with Upstream only and once with Downstream only.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[ShousenZHANG]

I'd like to take this one.

The cause is that _collect_relatives in airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py writes its results straight back into normal_task_ids / mapped_tasks_tuples. With both flags set, the upstream pass runs first and grows those sets, so the downstream pass then starts from the selected task's ancestors as well and picks up their other descendants — other in your example.

The fix is to run both directions from a snapshot of the originally requested tasks and take the union, which is what the dag_run_id is None branch in the same function already gets from dag.partial_subset(include_upstream=..., include_downstream=...). I'll add a regression test built from the Dag in your report that clears a with both flags and asserts other is not selected. PR shortly.