pierrejeambrun commented on code in PR #61550:
URL: https://github.com/apache/airflow/pull/61550#discussion_r3116493305
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1160,14 +1173,39 @@ def _create_orm_dagrun(
partition_key: str | None = None,
partition_date: datetime.datetime | None = None,
note: str | None = None,
+ bundle_version: str | None = None,
session: Session = NEW_SESSION,
) -> DagRun:
- bundle_version = None
+ resolved_bundle_version: str | None = None
+ use_resolved_dag = False
if not dag.disable_bundle_versioning:
- bundle_version = session.scalar(
- select(DagModel.bundle_version).where(DagModel.dag_id ==
dag.dag_id),
- )
- dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
+ if bundle_version is not None:
+ resolved_bundle_version = bundle_version
+ dag_version = DagVersion.get_latest_version(
+ dag.dag_id, bundle_version=resolved_bundle_version,
load_serialized_dag=True, session=session
+ )
+ if not dag_version:
+ raise DagVersionNotFound(
+ f"DAG with dag_id: '{dag.dag_id}' does not have a version
for bundle_version '{bundle_version}'"
+ )
+ use_resolved_dag = True
+ else:
+ resolved_bundle_version = session.scalar(
+ select(DagModel.bundle_version).where(DagModel.dag_id ==
dag.dag_id)
+ )
+ dag_version = DagVersion.get_latest_version(
+ dag.dag_id,
+ bundle_version=resolved_bundle_version,
+ load_serialized_dag=resolved_bundle_version is not None,
+ session=session,
+ )
+ if dag_version is None and resolved_bundle_version is not None:
+ dag_version = DagVersion.get_latest_version(dag.dag_id,
session=session)
+ elif dag_version is not None and resolved_bundle_version is not
None:
+ use_resolved_dag = True
Review Comment:
This branching is pretty subtle. The `elif dag_version is not None and
resolved_bundle_version is not None: use_resolved_dag = True` basically means
"even when the user didn't pass bundle_version, if DagModel has one recorded,
treat it as resolved and swap the dag." That's a behavior change from before
(previously `run.dag` stayed as the live dag regardless). Is that intentional I
don't think we should do that.
--
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]
