ephraimbuddy commented on code in PR #71459:
URL: https://github.com/apache/airflow/pull/71459#discussion_r4047333802
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -532,10 +532,7 @@ def dag_run_data(self) -> DRDataModel:
@property
def dag_versions(self) -> list[DagVersion]:
"""Return the DAG versions associated with the TIs of this DagRun."""
- # when the dag is in a versioned bundle, we keep the dag version fixed
Review Comment:
Removing this shortcut also makes patch_dag_run() load every TI/TIH when
serializing its response, even for a note-only update. Please use the DISTINCT
prefetch there too, after the mutations, and add a test asserting that
serialization leaves both collections unloaded.
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1388,16 +1374,142 @@ def
test_dag_run_dag_versions_with_null_created_dag_version(self, dag_maker, ses
EmptyOperator(task_id="empty")
dag_run = dag_maker.create_dagrun()
+ ti_version_ids = {ti.dag_version_id for ti in dag_run.task_instances
if ti.dag_version_id is not None}
+ assert ti_version_ids
+
dag_run.bundle_version = "some_bundle_version"
dag_run.created_dag_version_id = None
dag_run.created_dag_version = None
session.merge(dag_run)
session.flush()
- # This should return empty list, not [None]
- assert dag_run.dag_versions == []
- assert isinstance(dag_run.dag_versions, list)
- assert len(dag_run.dag_versions) == 0
+ # Derive from TI versions; never return [None] from a null
created_dag_version shortcut.
+ versions = dag_run.dag_versions
+ assert isinstance(versions, list)
+ assert {dv.id for dv in versions} == ti_version_ids
+ assert None not in versions
+
+ def
test_bundled_dag_run_dag_versions_after_partial_run_on_latest_version(self,
dag_maker, session):
+ """Bundled dag_versions include uncleared TI versions after a partial
run_on_latest_version clear.
+
+ Clearing a subset with run_on_latest_version=True bumps
created_dag_version_id
+ and bundle_version to latest while uncleared TIs stay on the old
version.
+ The property must report both.
+ """
+ with dag_maker(
+ "test_dag_run_dag_versions_partial_latest",
+ schedule=datetime.timedelta(days=1),
+ start_date=DEFAULT_DATE,
+ bundle_version="v1",
+ ):
+ EmptyOperator(task_id="0")
+ EmptyOperator(task_id="1")
+ dag_run = dag_maker.create_dagrun(state=State.RUNNING,
run_type=DagRunType.SCHEDULED)
+
+ old_dag_version = DagVersion.get_latest_version(dag_run.dag_id)
+ ti0, ti1 = sorted(dag_run.task_instances, key=lambda ti: ti.task_id)
+ for ti in (ti0, ti1):
+ ti.state = TaskInstanceState.SUCCESS
+ session.merge(ti)
+ dag_run.state = DagRunState.SUCCESS
+ session.merge(dag_run)
+ session.flush()
+
+ with dag_maker(
+ "test_dag_run_dag_versions_partial_latest",
+ schedule=datetime.timedelta(days=1),
+ start_date=DEFAULT_DATE,
+ bundle_version="v2",
+ ):
+ EmptyOperator(task_id="0")
+ EmptyOperator(task_id="1")
+ new_dag_version = DagVersion.get_latest_version(dag_run.dag_id)
+ assert old_dag_version.id != new_dag_version.id
+
+ clear_task_instances([ti0], session, run_on_latest_version=True)
+ session.commit()
+
+ dag_run = session.scalar(select(DagRun).where(DagRun.run_id ==
dag_run.run_id))
Review Comment:
Please include dag_id in this query and the equivalent query at line 1489,
or reload by the primary key. run_id is only unique within a Dag.
--
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]
