dstandish opened a new issue, #71455:
URL: https://github.com/apache/airflow/issues/71455
## 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
