You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hussein-awala (via GitHub)" <gi...@apache.org> on 2026/10/12 13:05:19 UTC
[PR] Update waiting tasks to the latest Dag version [airflow]
hussein-awala opened a new pull request, #74592:
URL: https://github.com/apache/airflow/pull/74592
<!-- SPDX-License-Identifier: Apache-2.0 -->
## Summary
Fix stale Dag-version associations on waiting task instances when an unversioned Dag changes during an active run.
For a local, unversioned bundle, a downstream task can start after its upstream task but still display an older Dag version. The upstream task's version reference is updated while the waiting downstream task's reference is left unchanged.
## Root cause
For runs without a pinned bundle version, `_verify_integrity_if_dag_changed()` updates unfinished task instances using:
```python
TI.state.in_(State.unfinished)
```
Although `State.unfinished` includes Python `None`, SQL `IN (..., NULL)` does not match rows whose state is `NULL`. Task instances waiting for their upstream dependencies can therefore be excluded from the version update, while running or deferred tasks receive the latest version.
Subsequent integrity checks can return early once the latest version is associated with any task instance or task-instance history record in the run, allowing the stale association to persist.
## Generic reproduction
1. Create an unversioned Dag containing `upstream >> downstream` and start a run.
2. Leave `upstream` running or deferred, with `downstream` still in `state=None`.
3. Update the Dag and persist a new serialized Dag version.
4. Run the scheduler's scheduling pass.
5. Observe that `upstream` references the new version but `downstream` retains the old one.
Expected: both unfinished task instances reference the new Dag version, including the downstream task before it becomes schedulable.
## Changes
- Explicitly include `TI.state.is_(None)` in the existing unfinished-task update filter.
- Add a database-backed, parametrized regression test for running and deferred upstream tasks.
- Verify that the downstream task receives the new version while waiting and retains it when scheduled after upstream success.
The fix preserves the existing Dag/run scope and versioned-bundle guard. It changes task-instance version metadata; it does not add source-file versioning to local bundles or retroactively repair previously inconsistent runs. A displayed Dag version alone is not proof of the source revision executed by a worker using an unversioned bundle.
## Validation
Reproduced on a local `main` checkout at `ef9feefd83`:
- Before the fix: both new regression cases failed at the downstream version assertion.
- After the fix: all five version-integrity tests passed through Breeze:
```bash
breeze run pytest airflow-core/tests/unit/jobs/test_scheduler_job.py -k verify_integrity -vs
```
The publication branch was then based on newer upstream `main`. Fast pre-commit checks, including the airflow-core mypy hook, and applicable manual checks passed for the two changed files. Ruff formatting/linting and `git diff --check` also passed.
**Draft validation limitation:** rerunning the tests in the newer-main worktree was blocked before pytest started by a local Breeze/Docker build-context mismatch. The broader test suites and documentation build listed by `breeze verify` have not been run. This draft does not claim a passing test rerun on the newer base; CI validation is still pending.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — pi (OpenAI GPT-6 Astra)
Generated-by: pi (OpenAI GPT-6 Astra) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
---
Drafted-by: pi (OpenAI GPT-6 Astra) (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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update waiting tasks to the latest Dag version [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #74592:
URL: https://github.com/apache/airflow/pull/74592
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update waiting tasks to the latest Dag version [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74592:
URL: https://github.com/apache/airflow/pull/74592#discussion_r4250047344
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3382,7 +3382,7 @@ def _verify_integrity_if_dag_changed(self, dag_run: DagRun, session: Session) ->
.where(
TI.dag_id == dag_run.dag_id,
TI.run_id == dag_run.run_id,
- TI.state.in_(State.unfinished),
+ or_(TI.state.in_(State.unfinished), TI.state.is_(None)),
Review Comment:
This looks like a regression from #60956, which replaced the Python loop (`if ti.state in State.unfinished`) with this bulk `UPDATE`. The `in` check matched `None`; SQL `IN (..., NULL)` never matches a NULL row. So 3.2.0 through 3.3.2 all have it. The `State.unfinished` filter at L3261 and the two in `taskinstance.py` already OR in `is_(None)`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update waiting tasks to the latest Dag version [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74592:
URL: https://github.com/apache/airflow/pull/74592#discussion_r4250047356
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -5957,6 +5957,49 @@ def test_verify_integrity_if_dag_changed(self, dag_maker):
session.rollback()
session.close()
+ @pytest.mark.parametrize("upstream_state", [State.RUNNING, State.DEFERRED])
Review Comment:
`RUNNING` and `DEFERRED` are both in `State.unfinished`, so the two cases go through the same path, and the state that matters here is the downstream TI's, which is `None` in both. Nothing checks the other side of the predicate either: if the `WHERE` matched every TI in the run, both cases would still pass. Could this drop the parametrize and add a finished TI (say a `SUCCESS` sibling) that asserts it keeps `original_version_id`?
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3382,7 +3382,7 @@ def _verify_integrity_if_dag_changed(self, dag_run: DagRun, session: Session) ->
.where(
TI.dag_id == dag_run.dag_id,
TI.run_id == dag_run.run_id,
- TI.state.in_(State.unfinished),
+ or_(TI.state.in_(State.unfinished), TI.state.is_(None)),
Review Comment:
This looks like a regression from #60956, which replaced the Python loop (`if ti.state in State.unfinished`) with this bulk `UPDATE`. The `in` check matched `None`; SQL `IN (..., NULL)` never matches a NULL row. So 3.2.0 through 3.3.2 all have it. Should this go on 3.3.3 too? The `State.unfinished` filter at L3261 and the two in `taskinstance.py` already OR in `is_(None)`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Update waiting tasks to the latest Dag version [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #74592:
URL: https://github.com/apache/airflow/pull/74592#discussion_r4251236326
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -5957,6 +5957,49 @@ def test_verify_integrity_if_dag_changed(self, dag_maker):
session.rollback()
session.close()
+ @pytest.mark.parametrize("upstream_state", [State.RUNNING, State.DEFERRED])
Review Comment:
I updated the test as suggested and tested it without and with the fix.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org