only_new clear can silently miss real new tasks after a prior "run on latest version" clear
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
## Problem
`_get_new_task_ids` (`airflow-core/src/airflow/models/taskinstance.py:270-309`), which powers the `only_new` clear feature added in #59764, computes "tasks added since this run's version" like this:
```python
# Use created_dag_version_id directly to get the DAG version the run was
# originally created with. We cannot use get_dag_for_run here because it
# falls back to the latest version when bundle_version is not set (e.g.
# LocalDagBundle), which would make current_dag == latest_dag and the diff
# always empty.
current_dag = None
if dag_run.created_dag_version_id:
current_dag = scheduler_dagbag.get_dag(version_id=dag_run.created_dag_version_id, session=session)
new_task_ids = set(latest_dag.task_ids) - set(current_dag.task_ids) if current_dag else set()
```
The comment explicitly documents the assumption that `created_dag_version_id` means "the version the run was originally created with." That assumption was correct when the field was immutable. Since #54984, `clear_task_instances(..., run_on_latest_version=True)` mutates `created_dag_version_id` to the latest version for a pinned run (while only bumping `dag_version_id` on the specific task instances that were cleared — see `taskinstance.py:477`). No new task-instance rows get created by that operation; it only updates the pointer and the already-existing, cleared TIs.
Once that has happened, `created_dag_version_id == latest`, so `current_dag == latest_dag`, and `new_task_ids` becomes `latest_dag.task_ids - latest_dag.task_ids` = the empty set — even if the run genuinely never got a task instance created for a task that was added between the run's true original version and latest.
## Repro
1. Create a pinned DagRun at version V1 with tasks A and B.
2. Deploy V2, which adds task C (V2 task_ids = {A, B, C}).
3. Clear task A with `run_on_latest_version=True`. Now `dag_run.created_dag_version_id == V2`; the run still has no TI for task C.
4. Run an `only_new` clear on this DagRun.
## Expected
Task C is identified as new and a TaskInstance is created for it.
## Actual
`_get_new_task_ids` returns an empty set (`current_dag` resolves to V2, same as `latest_dag`), so task C is silently never added, even though the run's actual task-instance rows never picked it up.
## Suggested fix
Determine "already has a TI for task X" by checking the run's actual `TaskInstance` rows directly (e.g. `{ti.task_id for ti in dag_run.get_task_instances(session=session)}`) rather than by diffing against a dag-version pointer (`created_dag_version_id`) that can no longer be trusted to represent "the version whose task set defines this run's existing task instances."
## Context
This is one concrete consequence of `created_dag_version_id`'s contract having drifted from "the version recorded at DagRun creation" (its documented meaning) to "the version this run should currently run at" (its behavior since #54984). See the companion issue tracking that broader contract problem: #71453
---
Drafted-by: Claude Code (Sonnet 5) (no human review before posting)
Contributor guide
Research direction
Start in airflow-core/src/airflow/models/taskinstance.py:270-309 at _get_new_task_ids, then read clear_task_instances around line 477 and trace how created_dag_version_id changes. Reproduce the V1/V2 sequence from the issue and verify that only_new identifies task C and creates its TaskInstance without relying on the mutated version pointer.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 52/100