pierrejeambrun commented on code in PR #61550:
URL: https://github.com/apache/airflow/pull/61550#discussion_r3116493298
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -485,6 +486,23 @@ def trigger_dag_run(
try:
dag = get_latest_version_of_dag(dag_bag, dag_id, session)
+
+ if body.bundle_version is not None:
+ if dag.disable_bundle_versioning:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"DAG with dag_id: '{dag_id}' does not support bundle
versioning",
+ )
+ dag_version = DagVersion.get_latest_version(
+ dag.dag_id, bundle_version=body.bundle_version,
load_serialized_dag=True, session=session
Review Comment:
We resolve the dag_version here, swap `dag`, then pass
`bundle_version=body.bundle_version` down to `create_dagrun` which re-fetches
the same `DagVersion` inside `_create_orm_dagrun`. That's one DB roundtrip we
could save — either pass the already-resolved `dag_version`/`dag` through, or
drop this block and let `_create_orm_dagrun` handle everything (which would
also make the asset materialize path consistent).
##########
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?
If so please add an inline comment explaining the reasoning, otherwise someone
will simplify this and break it.
##########
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,
Review Comment:
`_create_orm_dagrun` is also called from backfill paths, and they don't pass
`bundle_version`. But the new `elif` at line 1205 kicks in for those if
`DagModel.bundle_version` is set, which means backfill-created runs now get
`run.dag = resolved_dag` instead of the live dag. Subtle change — please add a
test confirming backfills still produce runs with the correct `dag_version_id`
and TIs.
##########
uv.lock:
##########
@@ -8753,17 +8753,17 @@ wheels = [
[[package]]
name = "aws-sam-translator"
Review Comment:
362 lines of churn here looks unrelated to this feature. Please revert or
split into its own commit with a clear reason unless some version bump is
actually required for the PR to work.
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -165,6 +166,9 @@ def _latest_version_select(
if load_bundle_model:
query = query.options(joinedload(cls.bundle))
Review Comment:
The `load_serialized_dag` kwarg is used inconsistently across callers —
`load_serialized_dag=True` on the explicit-bundle branch,
`load_serialized_dag=resolved_bundle_version is not None` on the auto-resolve
branch, and the fallback `get_latest_version(dag.dag_id, session=session)`
doesn't pass it at all. Worth auditing: we only actually touch
`dag_version.serialized_dag.dag` when `use_resolved_dag` is True, so any other
eager-load is wasted.
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -69,6 +69,17 @@
log = structlog.get_logger(__name__)
+_DAG_CALLBACK_ATTRS = (
+ "sla_miss_callback",
+ "on_success_callback",
+ "on_failure_callback",
+ "on_retry_callback",
+ "on_execute_callback",
+ "on_skipped_callback",
+ "has_on_success_callback",
+ "has_on_failure_callback",
Review Comment:
Worth a comment on this tuple explaining *why* we copy these specifically
(callbacks aren't serialized so we need to forward from the live dag) — so a
future reader doesn't wonder why we're not copying `max_active_runs`,
`catchup`, `params` etc. which by design come from the old serialized version.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -485,6 +486,23 @@ def trigger_dag_run(
try:
dag = get_latest_version_of_dag(dag_bag, dag_id, session)
+
+ if body.bundle_version is not None:
+ if dag.disable_bundle_versioning:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"DAG with dag_id: '{dag_id}' does not support bundle
versioning",
+ )
+ dag_version = DagVersion.get_latest_version(
+ dag.dag_id, bundle_version=body.bundle_version,
load_serialized_dag=True, session=session
+ )
+ if dag_version is None:
+ raise HTTPException(
+ status.HTTP_404_NOT_FOUND,
+ f"DAG with dag_id: '{dag_id}' does not have a version for
bundle_version '{body.bundle_version}'",
+ )
+ dag = dag_version.serialized_dag.dag
+
params = body.validate_context(dag)
Review Comment:
`validate_context` runs against the resolved (old) dag here, so
user-supplied conf is validated against the OLD version's param schema.
Probably the intended semantics (user asked for that specific old version), but
this is a behavior worth a dedicated test — trigger with an old bundle version
where param schema differs between versions, and assert validation uses the old
schema.
--
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]
