ephraimbuddy commented on code in PR #71696:
URL: https://github.com/apache/airflow/pull/71696#discussion_r3798505809
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -352,6 +352,28 @@ def _update_dagrun_to_latest_version(
session.flush()
+def _pin_versionless_tis_to_run_version(dag_run: DagRun, dag_version_id: UUID,
session: Session) -> None:
+ """
+ Give the run's unfinished task instances a dag version if they have none.
+
+ Once the run is pinned the scheduler stops backfilling versions onto them,
and one
+ without a version is never enqueued.
+ """
+ session.execute(
+ update(TaskInstance)
+ .where(
+ TaskInstance.dag_id == dag_run.dag_id,
+ TaskInstance.run_id == dag_run.run_id,
+ TaskInstance.dag_version_id.is_(None),
+ # State.unfinished holds None, which SQL IN never matches.
+ or_(TaskInstance.state.is_(None),
TaskInstance.state.in_(State.unfinished)),
+ )
+ .values(dag_version_id=dag_version_id)
+ .execution_options(synchronize_session=False)
+ )
+ session.expire(dag_run, ["task_instances"])
Review Comment:
```suggestion
```
This will fix the failing tests
--
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]
