You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "nathadfield (via GitHub)" <gi...@apache.org> on 2026/02/04 13:50:15 UTC
[PR] Add three-level run_on_latest_version configuration hierarchy [airflow]
nathadfield opened a new pull request, #61448:
URL: https://github.com/apache/airflow/pull/61448
## Description
Implements a three-level configuration hierarchy for `run_on_latest_version` to provide flexible control over bundle version selection when creating DAG runs.
This PR adds:
1. **Global configuration**: `[core] run_on_latest_version` option in `airflow.cfg`
2. **DAG-level configuration**: `@dag(run_on_latest_version=True/False)` parameter
3. **UI integration**: Clear/Rerun dialogs with version selection when applicable
The precedence hierarchy is: **DAG-level > Global config > System default (False)**
Fixes #60887
## Motivation
When working with bundle versioning in production, users need flexible control over which bundle version is used when creating DAG runs. Different DAGs may have different requirements:
- Some DAGs should always use the latest bundle version (e.g., data processing pipelines that need latest code)
- Others should use the version they were originally scheduled with (e.g., financial reporting that requires reproducibility)
Previously, this could only be controlled per-operation (trigger/clear). This PR enables configuration at the global and DAG levels for consistent behavior.
## Implementation Details
### Configuration Hierarchy
```python
# System default (if not configured anywhere)
run_on_latest_version = False
# Global configuration (applies to all DAGs unless overridden)
[core]
run_on_latest_version = True
# DAG-level configuration (highest precedence)
@dag(
dag_id="my_dag",
run_on_latest_version=True, # Overrides global config
...
)
def my_dag():
...
```
### Resolution Logic
The resolution logic in `SerializedDAG._resolve_bundle_version()` follows this precedence:
1. Check DAG-level `run_on_latest_version` parameter
2. If not set, check global `[core] run_on_latest_version` config
3. If not set, default to `False`
### UI Integration
Created a custom React hook (`useRunOnLatestVersion`) that:
- Fetches global config and DAG-level settings
- Determines the default checked state of the checkbox based on precedence hierarchy
- Properly handles nullable overrides to preserve the hierarchy
The Clear/Rerun dialogs show a "Run on latest version" checkbox when the current run's bundle version differs from the latest available version. The checkbox's default state (checked/unchecked) is determined by the configuration hierarchy.
https://github.com/user-attachments/assets/d0b4b24e-9c06-467a-b8bc-444a53cb955a
## Breaking Changes
None. The system default remains `False`, preserving existing behavior.
## Tests
```python
# Scheduler Exception Handling (tests/unit/jobs/test_scheduler_job.py)
test_scheduler_create_dag_runs_handles_bundle_version_unavailable
test_scheduler_create_dag_runs_asset_triggered_handles_bundle_version_unavailable
# DAG Model & Configuration Precedence (tests/unit/models/test_dagmodel.py)
test_create_dagrun_with_global_run_on_latest_version
test_create_dagrun_with_dag_level_run_on_latest_version
test_create_dagrun_disable_bundle_versioning_bypasses_logic
test_disable_bundle_versioning[True-some-version-None]
test_disable_bundle_versioning[False-some-version-some-version]
# API Endpoints (tests/unit/api_fastapi/)
test_dag_details_includes_run_on_latest_version # core_api/routes/public/test_dags.py
test_get_config_with_run_on_latest_version_true # core_api/routes/ui/test_config.py
test_get_config_with_run_on_latest_version_false # core_api/routes/ui/test_config.py
test_trigger_dag_run_bundle_version_not_yet_parsed # execution_api/versions/head/test_dag_runs.py
# Serialization (tests/unit/serialization/test_dag_serialization.py)
test_dag_run_on_latest_version_serialization[None]
test_dag_run_on_latest_version_serialization[True]
test_dag_run_on_latest_version_serialization[False]
```
## Documentation
Updated `docs/administration-and-deployment/dag-bundles.rst` with:
- Configuration options at all levels
- Precedence hierarchy explanation
- Examples of common use cases
- UI behavior description
## Related Issues/PRs
- Fixes #60887 - Global configuration for run_on_latest_version
- Related to #60880 - Bundle version support for TriggerDagRunOperator (will be addressed in separate PR)
## Generative AI Usage
**Was generative AI tooling used to co-author this PR?**
✅ **Yes**
**Generated-by:** Claude Sonnet 4.5 following the [Airflow contribution guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896261954
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,100 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
Review Comment:
Good work on this @nathadfield 🎉
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-3870282860
@jedcunningham Would you be able to give this a more detailed review when you get the time?
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2894617264
##########
airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx:
##########
@@ -245,7 +248,7 @@ const ClearTaskInstanceDialog = ({ onClose: onCloseDialog, open: openDialog, tas
open={open}
preventRunningTask={preventRunningTask}
/>
- ) : null}
+ ) : undefined}
Review Comment:
Already fixed with explicit triple-equals checks to satisfy ESLint's eqeqeq rule.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896232755
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,100 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
Review Comment:
Big enough change to add a newfragment for it too I feel: https://github.com/apache/airflow/tree/main/airflow-core/newsfragments
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4047109388
For now, I have extended the documentation to give greater clarity to the relationship between the different settings.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871552093
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1094,6 +1096,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
+ return _get_original_bundle_version(dag.dag_id, session)
+
+
+def _should_use_latest_version(
+ dag: SerializedDAG,
+) -> bool:
+ """
+ Determine whether to use latest bundle version based on precedence hierarchy.
+
+ Returns True if latest version should be used, False otherwise.
+ """
+ # Level 1: DAG-level configuration (explicit True or False)
+ if dag.run_on_latest_version is not None:
+ return dag.run_on_latest_version
+
+ # Level 2: Global configuration (fallback to False)
+ return airflow_conf.getboolean("core", "run_on_latest_version", fallback=False)
+
+
+def _get_config_source(
+ dag: SerializedDAG,
+) -> str:
+ """Return descriptive source of the bundle version configuration for logging."""
+ if dag.run_on_latest_version is not None:
+ return "DAG configuration"
+ if airflow_conf.getboolean("core", "run_on_latest_version", fallback=False):
+ return "global configuration"
+ return "system default"
+
+
+def _get_latest_bundle_version(dag_id: str, session: Session) -> str | None:
+ """
+ Get the latest bundle version from DagBundleModel, falling back to DagModel.
+
+ :param dag_id: The DAG ID
+ :param session: Database session
+ """
+ from airflow.models.dagbundle import DagBundleModel
+
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id))
+ if not dag_model:
+ log.warning(
+ "Cannot resolve latest bundle version: DagModel not found",
+ dag_id=dag_id,
+ )
+ return None
+
+ # Non-versioned bundle (e.g., LocalDagBundle) - use original version
+ if not dag_model.bundle_name:
Review Comment:
Thanks! Removed it.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871541956
##########
airflow-core/src/airflow/ui/src/components/Clear/useRunOnLatestVersion.ts:
##########
@@ -0,0 +1,93 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useState } from "react";
+
+import { useConfig } from "src/queries/useConfig";
+
+type UseRunOnLatestVersionProps = {
+ /**
+ * Current DAG bundle version
+ */
+ currentBundleVersion?: string | null;
+
+ /**
+ * DAG-level configuration from the DAG details.
+ * If defined (true or false), takes precedence over global config.
+ */
+ dagLevelConfig?: boolean | null;
+
+ /**
+ * DAG run bundle version
+ */
+ runBundleVersion?: string | null;
+};
+
+type UseRunOnLatestVersionResult = {
+ /**
+ * Setter for the checkbox value
+ */
+ setValue: (value: boolean) => void;
+
+ /**
+ * Whether the checkbox should be shown
+ */
+ shouldShowCheckbox: boolean;
+
+ /**
+ * Current value of the checkbox (controlled)
+ */
+ value: boolean;
+};
+
+/**
+ * Custom hook for managing "Run on Latest Version" checkbox state.
+ *
+ * Implements the three-level precedence hierarchy:
+ * 1. DAG-level configuration (highest priority)
+ * 2. Global configuration
+ * 3. System default (false)
+ *
+ * Uses nullable override pattern: state is null until user interacts,
+ * avoiding the useState + useEffect synchronization anti-pattern.
+ */
+export const useRunOnLatestVersion = ({
+ currentBundleVersion,
+ dagLevelConfig,
+ runBundleVersion,
+}: UseRunOnLatestVersionProps): UseRunOnLatestVersionResult => {
+ const globalConfigValue = useConfig("run_on_latest_version");
+ const globalDefault = globalConfigValue === "true";
Review Comment:
Good catch! Fixed to `Boolean(globalConfigValue)`
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2889679008
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,96 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
+--------------------------------------------
+
+When using versioned DAG bundles (such as ``GitDagBundle``), Airflow provides flexible control
+over which bundle version is used for DAG runs through a three-level configuration hierarchy.
+
+.. note::
+
+ Bundle versioning is only supported by certain bundle types (like ``GitDagBundle``). Local bundles
+ (``LocalDagBundle``) do not support versioning and will always use the latest code.
+
+Precedence Hierarchy
+~~~~~~~~~~~~~~~~~~~~
+
+The bundle version used for a DAG run is determined by the following precedence (highest to lowest):
+
+1. **DAG Level**: The DAG's ``run_on_latest_version`` parameter
+2. **Global Level**: The ``[core] run_on_latest_version`` configuration setting
+3. **System Default**: Use the original bundle version (``False`` - for reproducibility)
+
+Global Configuration
+~~~~~~~~~~~~~~~~~~~~
+
+You can set organization-wide defaults using the ``[core] run_on_latest_version`` configuration option.
+This setting determines the default behavior for all DAGs unless overridden at the DAG level.
+
+.. code-block:: ini
+
+ [core]
+ run_on_latest_version = False # Default - use original bundle version
+ # run_on_latest_version = True # Alternative - use latest bundle version
+
+- When **False** (default): DAG runs use the original bundle version that was active when the DAG was first parsed, ensuring reproducibility and consistency across the entire run.
+- When **True**: DAG runs use the latest available bundle version when triggered, rerun, or cleared, ensuring DAGs always run with the most up-to-date code.
Review Comment:
I've rewritten the docs section to clarify. See the reply to your general comment for the full explanation.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2889677419
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,96 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
+--------------------------------------------
+
+When using versioned DAG bundles (such as ``GitDagBundle``), Airflow provides flexible control
+over which bundle version is used for DAG runs through a three-level configuration hierarchy.
+
+.. note::
+
+ Bundle versioning is only supported by certain bundle types (like ``GitDagBundle``). Local bundles
+ (``LocalDagBundle``) do not support versioning and will always use the latest code.
+
+Precedence Hierarchy
+~~~~~~~~~~~~~~~~~~~~
+
+The bundle version used for a DAG run is determined by the following precedence (highest to lowest):
+
+1. **DAG Level**: The DAG's ``run_on_latest_version`` parameter
+2. **Global Level**: The ``[core] run_on_latest_version`` configuration setting
+3. **System Default**: Use the original bundle version (``False`` - for reproducibility)
Review Comment:
The system default is just False, hardcoded as the fallback when neither the DAG-level parameter nor the global config option is set. It's not user-configurable. I've updated the docs to make this clearer by removing the "system default" label and just listing Default: False in the precedence.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871555861
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1094,6 +1096,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
+ return _get_original_bundle_version(dag.dag_id, session)
+
+
+def _should_use_latest_version(
+ dag: SerializedDAG,
+) -> bool:
+ """
+ Determine whether to use latest bundle version based on precedence hierarchy.
+
+ Returns True if latest version should be used, False otherwise.
+ """
+ # Level 1: DAG-level configuration (explicit True or False)
+ if dag.run_on_latest_version is not None:
+ return dag.run_on_latest_version
+
+ # Level 2: Global configuration (fallback to False)
+ return airflow_conf.getboolean("core", "run_on_latest_version", fallback=False)
+
+
+def _get_config_source(
+ dag: SerializedDAG,
+) -> str:
+ """Return descriptive source of the bundle version configuration for logging."""
+ if dag.run_on_latest_version is not None:
+ return "DAG configuration"
+ if airflow_conf.getboolean("core", "run_on_latest_version", fallback=False):
+ return "global configuration"
+ return "system default"
+
+
+def _get_latest_bundle_version(dag_id: str, session: Session) -> str | None:
+ """
+ Get the latest bundle version from DagBundleModel, falling back to DagModel.
+
+ :param dag_id: The DAG ID
+ :param session: Database session
+ """
+ from airflow.models.dagbundle import DagBundleModel
+
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id))
+ if not dag_model:
+ log.warning(
+ "Cannot resolve latest bundle version: DagModel not found",
+ dag_id=dag_id,
+ )
+ return None
+
+ # Non-versioned bundle (e.g., LocalDagBundle) - use original version
+ if not dag_model.bundle_name:
+ log.debug(
+ "No bundle_name configured, using original version",
+ dag_id=dag_id,
+ bundle_version=dag_model.bundle_version,
+ )
+ return dag_model.bundle_version
+
+ dag_bundle = session.scalar(select(DagBundleModel).where(DagBundleModel.name == dag_model.bundle_name))
+ if not dag_bundle:
+ log.warning(
+ "DagBundleModel not found, falling back to original version",
+ dag_id=dag_id,
+ bundle_name=dag_model.bundle_name,
+ fallback_version=dag_model.bundle_version,
+ )
+ return dag_model.bundle_version
+
+ log.debug(
+ "Resolved latest bundle version",
+ dag_id=dag_id,
+ bundle_name=dag_model.bundle_name,
+ bundle_version=dag_bundle.version,
+ )
+ return dag_bundle.version
+
+
+def _get_original_bundle_version(dag_id: str, session: Session) -> str | None:
+ """Get the original bundle version from DagModel."""
+ version = session.scalar(select(DagModel.bundle_version).where(DagModel.dag_id == dag_id))
Review Comment:
Good point! Replaced the dead bundle_name check with an explicit `if dag_bundle.version is None:` check that falls back to `dag_model.bundle_version`. Added a test confirming non-versioned bundles use the original version even when `run_on_latest_version=True`.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-3856862932
My immediate reaction is, not sure hows its much different than the existing [[dag_processor] disable_bundle_versioning](https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#disable-bundle-versioning) and `disable_bundle_versioning` on Dag. My 30s review for the day :)
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896200595
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,46 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
)
- )
- dag_run.consumed_asset_events.extend(asset_events)
- session.flush()
- apdr.created_dag_run_id = dag_run.id
- session.flush()
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except Exception:
Review Comment:
Fair point! I've switched to `except DBAPIError: raise` to match the full retry contract of `@retry_db_transaction` and updated in both locations.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-3855963780
Calling our resident Dag Version experts: @ephraimbuddy @jedcunningham
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4014256479
@ephraimbuddy Can you take a look at it too if you have any concerns?
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4069208677
@jedcunningham Just to address some of your other comments.
The `bundle_version` on DagRun does get overwritten on clear when `run_on_latest_version=True`, but that's the intended opt-in behavior. When `False` (the default), the original version is preserved and tasks rerun with the same code. The version history matters because it determines which DagVersion (and therefore which serialized DAG code) the tasks actually execute with.
On using `DagModel.bundle_version` vs `DagBundleModel.version`, I agree the race window is a real concern, but there's a trade-off worth discussing.
If we use `DagModel.bundle_version`, there's a window after bundle refresh where a user clears a run with "run on latest version" checked but silently gets the previous version because the DAG hasn't been re-parsed yet. They asked for latest but got stale code with no indication.
The current approach (`DagBundleModel.version`) throws `BundleVersionUnavailable` in that same window. It's noisier but at least the user knows they didn't get what they asked for and can retry once parsing completes.
One option is a middle ground: use `DagModel.bundle_version` but log a warning when it differs from `DagBundleModel.version`, so there's visibility that a newer version exists but hasn't been parsed yet. Or we could surface it in the API response.
What's your preference? Silent fallback, error, or warn-and-fallback?
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4005650709
For additional context, in Airflow 2.x bundle versioning didn't exist, so the behavior was always to run with the latest code. Airflow 3.x introduced bundle versioning, which is a good improvement, but the default changed to running with the version as it was at parse time. This PR just gives teams the ability to control which of those two behaviors is the default, both globally and per-DAG. Some teams will prefer the Airflow 2 behavior (always latest), others will want the reproducibility that versioning provides. I don't think it's an unreasonable request to have that be configurable rather than hardcoded.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871550109
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py:
##########
@@ -133,6 +133,18 @@ def trigger_dag_run(
"message": f"A run already exists for Dag '{dag_id}' with run_id '{run_id}'",
},
)
+ except BundleVersionUnavailable as e:
Review Comment:
Fixed!. Added a `BundleVersionUnavailable` catch that returns HTTP 503, matching the pattern in `execution_api/routes/dag_runs.py`. I also added a test confirming the 503 response.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2894610252
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1097,6 +1099,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
Review Comment:
This codebase uses structlog, which natively supports keyword arguments. The `self.log.warning("msg", dag_id=dag_id)` pattern is idiomatic so no % formatting needed.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2895979632
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,44 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
+ )
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except OperationalError:
Review Comment:
Thanks for the quick follow-up here. One remaining concern: re-raising only `OperationalError` is still narrower than the retry contract for `@retry_db_transaction`, which is based on `DBAPIError`. Other `DBAPIError` subclasses (e.g. `IntegrityError`) can still be caught by the broad `except Exception` and get swallowed. Could we switch this to `except DBAPIError: raise` (or rollback + re-raise), then keep non-DB exceptions as log-and-continue? Same applies to the analogous block in `_create_dag_runs_asset_triggered`.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896244445
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,100 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
Review Comment:
Sure, I can write one up.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4012733943
@kaxil I think that everything. I saw a test failed but I don't think it's related to anything I've done.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2894605333
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,46 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
)
- )
- dag_run.consumed_asset_events.extend(asset_events)
- session.flush()
- apdr.created_dag_run_id = dag_run.id
- session.flush()
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except Exception:
Review Comment:
Good catch. I've added except `OperationalError: raise` before each except Exception block so DB errors propagate correctly. Fixed in the latest push.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896279761
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1245,6 +1343,112 @@ def test_dag_add_task_sets_default_task_group(self):
assert task_group.get_child_by_label("task_with_task_group") == task_with_task_group
assert dag.get_task("task_group.task_with_task_group") == task_with_task_group
+ def test_create_dagrun_system_default_uses_original_version(self, dag_maker, session):
+ """Test system default: when DAG=None and global=False, uses original bundle version."""
+ with conf_vars({("core", "run_on_latest_version"): "False"}):
+ with dag_maker("test_system_default", run_on_latest_version=None):
+ ...
+
+ self._setup_bundle(session, "test_system_default", "test_bundle_system_default")
+
+ dr = self._create_test_dagrun(dag_maker, run_id="test_system_default")
+ assert dr.bundle_version == "v1.0.0"
+
+ def test_create_dagrun_disable_bundle_versioning_bypasses_logic(self, dag_maker, session):
+ """Test that disable_bundle_versioning=True bypasses all bundle version logic."""
+ with conf_vars({("core", "run_on_latest_version"): "True"}):
+ with dag_maker("test_bypass", disable_bundle_versioning=True):
+ ...
+
+ self._setup_bundle(session, "test_bypass", "test_bundle_bypass")
+
+ dr = self._create_test_dagrun(dag_maker, run_id="test_bypass")
+ assert dr.bundle_version is None
+
+ def test_create_dagrun_race_condition_fails_fast(self, dag_maker, session):
+ """Test that race condition (bundle updated but not serialized) raises clear error."""
+ from airflow.models.dag_version import DagVersion
+ from airflow.models.dagbundle import DagBundleModel
Review Comment:
These imports are inside the test function body. Please move them to the module-level imports (DagBundleModel is already imported at the top) unless there is a specific need for a local import.
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1245,6 +1343,112 @@ def test_dag_add_task_sets_default_task_group(self):
assert task_group.get_child_by_label("task_with_task_group") == task_with_task_group
assert dag.get_task("task_group.task_with_task_group") == task_with_task_group
+ def test_create_dagrun_system_default_uses_original_version(self, dag_maker, session):
+ """Test system default: when DAG=None and global=False, uses original bundle version."""
+ with conf_vars({("core", "run_on_latest_version"): "False"}):
+ with dag_maker("test_system_default", run_on_latest_version=None):
+ ...
+
+ self._setup_bundle(session, "test_system_default", "test_bundle_system_default")
+
+ dr = self._create_test_dagrun(dag_maker, run_id="test_system_default")
+ assert dr.bundle_version == "v1.0.0"
+
+ def test_create_dagrun_disable_bundle_versioning_bypasses_logic(self, dag_maker, session):
+ """Test that disable_bundle_versioning=True bypasses all bundle version logic."""
+ with conf_vars({("core", "run_on_latest_version"): "True"}):
+ with dag_maker("test_bypass", disable_bundle_versioning=True):
+ ...
+
+ self._setup_bundle(session, "test_bypass", "test_bundle_bypass")
+
+ dr = self._create_test_dagrun(dag_maker, run_id="test_bypass")
+ assert dr.bundle_version is None
+
+ def test_create_dagrun_race_condition_fails_fast(self, dag_maker, session):
+ """Test that race condition (bundle updated but not serialized) raises clear error."""
+ from airflow.models.dag_version import DagVersion
+ from airflow.models.dagbundle import DagBundleModel
+
+ with dag_maker("test_race_condition", run_on_latest_version=True):
+ ...
+
+ # Create DagBundleModel with v2.0.0 (simulating bundle refresh)
+ dag_bundle = DagBundleModel(name="test_bundle_race", version="v2.0.0")
+ session.add(dag_bundle)
+ session.flush()
+
+ # Update DagModel to point to this bundle
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == "test_race_condition"))
+ dag_model.bundle_name = "test_bundle_race"
+ dag_model.bundle_version = "v1.0.0" # Original version
+ session.flush()
+
+ # Create only ONE DagVersion for v1.0.0 (v2.0.0 not serialized yet - simulating race)
+ existing_dag_version = session.scalar(
+ select(DagVersion).where(DagVersion.dag_id == "test_race_condition").limit(1)
+ )
+ if existing_dag_version:
+ existing_dag_version.bundle_name = "test_bundle_race"
+ existing_dag_version.bundle_version = "v1.0.0"
+ session.flush()
+
+ # Request "run on latest" - should get v2.0.0 from DagBundleModel
+ # But DagVersion for v2.0.0 doesn't exist yet (race condition)
+ # Should raise exception with clear message about temporary condition
+ with pytest.raises(
+ BundleVersionUnavailable,
+ match="Cannot create DagRun.*bundle version v2.0.0.*not been parsed yet.*retry",
+ ):
+ self._create_test_dagrun(dag_maker, run_id="test_race_fail")
+
+ @mock.patch("airflow.configuration.conf.getboolean")
+ def test_create_dagrun_dag_none_uses_global_true(self, mock_getboolean, dag_maker, session):
+ """Test DAG=None with Global=True uses latest version (not original)."""
+ mock_getboolean.side_effect = lambda section, key, fallback=None: (
+ True if section == "core" and key == "run_on_latest_version" else fallback
+ )
+
+ # DAG level is None (not explicitly set) - should use global True
+ with dag_maker("test_dag_none_global_true", run_on_latest_version=None):
+ ...
+
+ self._setup_bundle(session, "test_dag_none_global_true", "test_bundle_none_global")
+
+ dr = self._create_test_dagrun(dag_maker)
+ # Should use v2.0.0 because global=True and DAG=None falls through
+ assert dr.bundle_version == "v2.0.0"
+
+ def test_create_dagrun_non_versioned_bundle_uses_original_version(self, dag_maker, session):
+ """Test that non-versioned bundles (e.g. LocalDagBundle) fall back to original version."""
+ from airflow.models.dag_version import DagVersion
+ from airflow.models.dagbundle import DagBundleModel
Review Comment:
These imports are inside the test function body. Please move them to the module-level imports (DagBundleModel is already imported at the top) unless there is a specific need for a local import.
##########
airflow-core/tests/unit/models/test_backfill.py:
##########
@@ -162,6 +162,8 @@ def test_create_backfill_clear_existing_bundle_version(dag_maker, session, run_o
"""
Verify that when backfill clears an existing dag run, bundle version is cleared.
"""
+ from airflow.models.dag_version import DagVersion
+
Review Comment:
This local import inside the test body is avoidable and makes dependencies less discoverable. Please move DagVersion to the module-level imports unless the local import is needed for circular import avoidance or test isolation.
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -3542,6 +3746,7 @@ def test_validate_setup_teardown_trigger_rule(self):
)
def test_disable_bundle_versioning(disable, bundle_version, expected, dag_maker, session, clear_dags):
"""When bundle versioning is disabled for a dag, the dag run should not have a bundle version."""
+ from airflow.models.dag_version import DagVersion
Review Comment:
Importing DagVersion inside the test function body makes imports harder to track and is inconsistent with the module-level import style used elsewhere in this file. Please move this to the top-level imports unless there's a specific reason for a local import.
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1232,6 +1232,104 @@ def test_create_dagrun_partition_key(self, partition_key, dag_maker):
)
assert dr.partition_key == partition_key
+ def _setup_bundle(self, session, dag_id, bundle_name, original_version="v1.0.0", latest_version="v2.0.0"):
+ """
+ Set up bundle configuration for testing bundle version resolution.
+
+ Creates a DagBundleModel with the latest version, updates the DagModel with bundle info,
+ and creates DagVersion entries for both original and latest versions to simulate
+ what happens when different bundle versions are parsed.
+ """
+ from airflow.models.dag_version import DagVersion
+ from airflow.models.dagbundle import DagBundleModel
Review Comment:
These imports are inside a helper method body. To keep imports consistent and avoid repeated imports during test runs, please move them to the module import section unless there is a specific need for a local import (e.g. circular import avoidance).
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py:
##########
@@ -143,6 +143,18 @@ def trigger_dag_run(
"message": f"A run already exists for Dag '{dag_id}' with run_id '{run_id}'",
},
)
+ except BundleVersionUnavailable as e:
+ log.warning(
+ "Bundle version unavailable when triggering DAG run",
Review Comment:
The warning log for BundleVersionUnavailable omits the exception detail, which makes production troubleshooting harder. Consider including the exception message in the log context (e.g. add it to `extra` or include it in the formatted message).
```suggestion
"Bundle version unavailable when triggering DAG run: %s",
e,
```
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1097,6 +1099,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
+ return _get_original_bundle_version(dag.dag_id, session)
+
+
+def _should_use_latest_version(
+ dag: SerializedDAG,
+) -> bool:
+ """
+ Determine whether to use latest bundle version based on precedence hierarchy.
+
+ Returns True if latest version should be used, False otherwise.
+ """
+ # Level 1: DAG-level configuration (explicit True or False)
+ if dag.run_on_latest_version is not None:
+ return dag.run_on_latest_version
+
+ # Level 2: Global configuration (fallback to False)
+ return airflow_conf.getboolean("core", "run_on_latest_version", fallback=False)
+
+
+def _get_config_source(
+ dag: SerializedDAG,
+) -> str:
+ """Return descriptive source of the bundle version configuration for logging."""
+ if dag.run_on_latest_version is not None:
+ return "DAG configuration"
+ if airflow_conf.has_option("core", "run_on_latest_version"):
+ return "global configuration"
+ return "system default"
+
+
+def _get_latest_bundle_version(dag_id: str, session: Session) -> str | None:
+ """
+ Get the latest bundle version from DagBundleModel, falling back to DagModel.
+
+ :param dag_id: The DAG ID
+ :param session: Database session
+ """
+ from airflow.models.dagbundle import DagBundleModel
+
Review Comment:
Importing DagBundleModel inside _get_latest_bundle_version() violates the project rule to keep imports at module scope. Please move this import to the top of the file (or add a short comment explaining why a local import is required, e.g. circular import avoidance).
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2889234673
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,96 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
+--------------------------------------------
+
+When using versioned DAG bundles (such as ``GitDagBundle``), Airflow provides flexible control
+over which bundle version is used for DAG runs through a three-level configuration hierarchy.
+
+.. note::
+
+ Bundle versioning is only supported by certain bundle types (like ``GitDagBundle``). Local bundles
+ (``LocalDagBundle``) do not support versioning and will always use the latest code.
+
+Precedence Hierarchy
+~~~~~~~~~~~~~~~~~~~~
+
+The bundle version used for a DAG run is determined by the following precedence (highest to lowest):
+
+1. **DAG Level**: The DAG's ``run_on_latest_version`` parameter
+2. **Global Level**: The ``[core] run_on_latest_version`` configuration setting
+3. **System Default**: Use the original bundle version (``False`` - for reproducibility)
Review Comment:
What does the system default mean? How is it set?
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,96 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
+--------------------------------------------
+
+When using versioned DAG bundles (such as ``GitDagBundle``), Airflow provides flexible control
+over which bundle version is used for DAG runs through a three-level configuration hierarchy.
+
+.. note::
+
+ Bundle versioning is only supported by certain bundle types (like ``GitDagBundle``). Local bundles
+ (``LocalDagBundle``) do not support versioning and will always use the latest code.
+
+Precedence Hierarchy
+~~~~~~~~~~~~~~~~~~~~
+
+The bundle version used for a DAG run is determined by the following precedence (highest to lowest):
+
+1. **DAG Level**: The DAG's ``run_on_latest_version`` parameter
+2. **Global Level**: The ``[core] run_on_latest_version`` configuration setting
+3. **System Default**: Use the original bundle version (``False`` - for reproducibility)
+
+Global Configuration
+~~~~~~~~~~~~~~~~~~~~
+
+You can set organization-wide defaults using the ``[core] run_on_latest_version`` configuration option.
+This setting determines the default behavior for all DAGs unless overridden at the DAG level.
+
+.. code-block:: ini
+
+ [core]
+ run_on_latest_version = False # Default - use original bundle version
+ # run_on_latest_version = True # Alternative - use latest bundle version
+
+- When **False** (default): DAG runs use the original bundle version that was active when the DAG was first parsed, ensuring reproducibility and consistency across the entire run.
+- When **True**: DAG runs use the latest available bundle version when triggered, rerun, or cleared, ensuring DAGs always run with the most up-to-date code.
Review Comment:
These definitions are not clear. Are you defining what happens when a dagrun is cleared? When creating a dagrun, there's no old bundle version because the dagrun is _about_ to be created( didn't exist before) so it's always created with the latest bundle version. So if a user set this to False or True, it's unclear what new dagrun being created should be created with and how to determine an old bundle in this situation
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4037485746
I guess I can see three alternatives.
**Option A: Consolidate into an enum**
Replace both `disable_bundle_versioning` and `run_on_latest_version` with a single config like `bundle_version_strategy` with values:
- **disabled**: no versioning at all (`current disable_bundle_versioning=True`)
- **pinned**: use the version from when the run was created (current default)
- **latest**: always use the latest version on clear/rerun (what `run_on_latest_version=True` does)
This addresses the overlap concern directly and is arguably cleaner but has a big impact.
**Option B: Extend disable_bundle_versioning**
Rename/extend the existing flag to support a third state. E.g., `bundle_versioning = disabled | pinned | latest`.
**Option C: Keep both, improve docs**
Keep the current approach but add clear documentation explaining the distinction: `disable_bundle_versioning` turns off version tracking entirely (no version recorded), while `run_on_latest_version` keeps version tracking but controls the default rerun behavior. Add a comparison table.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-3984574411
@ephraimbuddy @kaxil Thanks for the review. All points addressed and tests running successfully.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2892798549
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -494,6 +495,11 @@ def trigger_dag_run(
current_user_id = user.get_id()
dag_run.note = (dag_run_note, current_user_id)
return dag_run
+ except BundleVersionUnavailable as e:
+ raise HTTPException(
+ status.HTTP_503_SERVICE_UNAVAILABLE,
+ f"Bundle version not yet available. Please retry: {e}",
Review Comment:
This endpoint returns a plain-string `detail` for `BundleVersionUnavailable`, while the execution API returns a structured `{reason, message}` object for the analogous 503. For client consistency, consider returning a structured `detail` here too (e.g. include a stable `reason` key) so UI/SDK consumers can reliably detect and handle this retryable condition.
```suggestion
{
"reason": "BUNDLE_VERSION_UNAVAILABLE",
"message": f"Bundle version not yet available. Please retry: {e}",
},
```
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1097,6 +1099,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
+ return _get_original_bundle_version(dag.dag_id, session)
+
+
+def _should_use_latest_version(
+ dag: SerializedDAG,
+) -> bool:
+ """
+ Determine whether to use latest bundle version based on precedence hierarchy.
+
+ Returns True if latest version should be used, False otherwise.
+ """
+ # Level 1: DAG-level configuration (explicit True or False)
+ if dag.run_on_latest_version is not None:
+ return dag.run_on_latest_version
+
+ # Level 2: Global configuration (fallback to False)
+ return airflow_conf.getboolean("core", "run_on_latest_version", fallback=False)
+
+
+def _get_config_source(
+ dag: SerializedDAG,
+) -> str:
+ """Return descriptive source of the bundle version configuration for logging."""
+ if dag.run_on_latest_version is not None:
+ return "DAG configuration"
+ if airflow_conf.has_option("core", "run_on_latest_version"):
+ return "global configuration"
+ return "system default"
+
+
+def _get_latest_bundle_version(dag_id: str, session: Session) -> str | None:
+ """
+ Get the latest bundle version from DagBundleModel, falling back to DagModel.
+
+ :param dag_id: The DAG ID
+ :param session: Database session
+ """
+ from airflow.models.dagbundle import DagBundleModel
+
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id))
+ if not dag_model:
+ log.warning(
+ "Cannot resolve latest bundle version: DagModel not found",
+ dag_id=dag_id,
+ )
Review Comment:
These `log.debug(..., dag_id=..., source=...)` / `log.warning(..., dag_id=...)` calls pass arbitrary keyword arguments, which will raise `TypeError` with standard `logging.Logger`. Use `%s` formatting args or `extra={...}` instead (consistent with other logging in this file that uses printf-style args).
##########
airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx:
##########
@@ -245,7 +248,7 @@ const ClearTaskInstanceDialog = ({ onClose: onCloseDialog, open: openDialog, tas
open={open}
preventRunningTask={preventRunningTask}
/>
- ) : null}
+ ) : undefined}
Review Comment:
In JSX, returning `null` is the conventional way to render nothing; returning `undefined` can trip linting rules and is less idiomatic. Prefer `) : null}` here for consistency with other components.
```suggestion
) : null}
```
##########
airflow-core/src/airflow/ui/src/components/Clear/useRunOnLatestVersion.ts:
##########
@@ -0,0 +1,94 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useState } from "react";
+
+import { useConfig } from "src/queries/useConfig";
+
+type UseRunOnLatestVersionProps = {
+ /**
+ * Current DAG bundle version
+ */
+ currentBundleVersion?: string | null;
+
+ /**
+ * DAG-level configuration from the DAG details.
+ * If defined (true or false), takes precedence over global config.
+ */
+ dagLevelConfig?: boolean | null;
+
+ /**
+ * DAG run bundle version
+ */
+ runBundleVersion?: string | null;
+};
+
+type UseRunOnLatestVersionResult = {
+ /**
+ * Setter for the checkbox value
+ */
+ setValue: (value: boolean) => void;
+
+ /**
+ * Whether the checkbox should be shown
+ */
+ shouldShowCheckbox: boolean;
+
+ /**
+ * Current value of the checkbox (controlled)
+ */
+ value: boolean;
+};
+
+/**
+ * Custom hook for managing "Run on Latest Version" checkbox state.
+ *
+ * Implements the three-level precedence hierarchy:
+ * 1. DAG-level configuration (highest priority)
+ * 2. Global configuration
+ * 3. System default (false)
+ *
+ * Uses nullable override pattern: state is null until user interacts,
+ * avoiding the useState + useEffect synchronization anti-pattern.
+ */
+export const useRunOnLatestVersion = ({
+ currentBundleVersion,
+ dagLevelConfig,
+ runBundleVersion,
+}: UseRunOnLatestVersionProps): UseRunOnLatestVersionResult => {
+ const globalConfigValue = useConfig("run_on_latest_version");
+ const globalDefault = Boolean(globalConfigValue);
+
+ // Precedence: DAG-level > Global > System default (false)
+ const defaultValue = dagLevelConfig ?? globalDefault;
+
+ // Nullable override: null until user interacts, then stores their choice
+ const [userOverride, setUserOverride] = useState<boolean | null>(null);
+ const value = userOverride ?? defaultValue;
+
+ // Show checkbox only when versions differ and run has a valid bundle version
+ const hasValidRunVersion =
+ runBundleVersion !== undefined && runBundleVersion !== null && runBundleVersion !== "";
+ const shouldShowCheckbox = hasValidRunVersion && currentBundleVersion !== runBundleVersion;
Review Comment:
As written, the checkbox will render when `currentBundleVersion` is still `undefined` (e.g. DAG details not loaded yet) because `undefined !== runBundleVersion` is `true`. This can cause the checkbox to flash or show with an incorrect comparison. Require a valid `currentBundleVersion` (non-empty string) before comparing, or delay showing until DAG details have loaded.
```suggestion
// Show checkbox only when both versions are valid and differ
const hasValidRunVersion =
runBundleVersion !== undefined && runBundleVersion !== null && runBundleVersion !== "";
const hasValidCurrentVersion =
currentBundleVersion !== undefined &&
currentBundleVersion !== null &&
currentBundleVersion !== "";
const shouldShowCheckbox =
hasValidRunVersion && hasValidCurrentVersion && currentBundleVersion !== runBundleVersion;
```
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1097,6 +1099,116 @@ def get_edge_info(self, upstream_task_id: str, downstream_task_id: str) -> EdgeI
return empty
+def _resolve_bundle_version(
+ dag: SerializedDAG,
+ session: Session,
+) -> str | None:
+ """
+ Resolve the bundle version for a DAG run based on the precedence hierarchy.
+
+ Precedence (highest to lowest):
+ 1. DAG-level configuration (dag.run_on_latest_version)
+ 2. Global configuration (core.run_on_latest_version)
+ 3. System default (use original version from DagModel)
+
+ :param dag: The serialized DAG
+ :param session: Database session
+ :return: The resolved bundle version, or None if versioning is disabled
+ """
+ if dag.disable_bundle_versioning:
+ return None
+
+ # Determine whether to use latest version based on precedence hierarchy
+ use_latest = _should_use_latest_version(dag)
+ source = _get_config_source(dag)
+
+ if use_latest:
+ log.debug("Using latest bundle version", dag_id=dag.dag_id, source=source)
+ return _get_latest_bundle_version(dag.dag_id, session)
+
+ log.debug("Using original bundle version", dag_id=dag.dag_id, source=source)
Review Comment:
These `log.debug(..., dag_id=..., source=...)` / `log.warning(..., dag_id=...)` calls pass arbitrary keyword arguments, which will raise `TypeError` with standard `logging.Logger`. Use `%s` formatting args or `extra={...}` instead (consistent with other logging in this file that uses printf-style args).
```suggestion
log.debug("Using latest bundle version: dag_id=%s, source=%s", dag.dag_id, source)
return _get_latest_bundle_version(dag.dag_id, session)
log.debug("Using original bundle version: dag_id=%s, source=%s", dag.dag_id, source)
```
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896294577
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,100 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
Review Comment:
detailed might be better so it look like https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#serialization-interface-changes
Could be just a handful of lines that cross link to this dag bundles sectino
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896267189
##########
airflow-core/docs/administration-and-deployment/dag-bundles.rst:
##########
@@ -139,6 +139,100 @@ are configured so that impersonated users can access bundle files created by the
the need for shared group permissions.
+Configuring Default Bundle Version Behavior
Review Comment:
Would you want a feature one-liner or a more detailed significant doc?
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2892559185
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,46 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
)
- )
- dag_run.consumed_asset_events.extend(asset_events)
- session.flush()
- apdr.created_dag_run_id = dag_run.id
- session.flush()
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except Exception:
Review Comment:
Everything else in this PR looks good to me, apart from this major blocker: catching broad `Exception` here can swallow `DBAPIError` and prevent the `@retry_db_transaction` rollback/retry path from kicking in. That can leave the session in a failed state and skip retries for transient DB failures. Could we either (1) add an explicit `except DBAPIError: session.rollback(); raise` before this block, or (2) rollback + re-raise DB errors, and keep only the non-DB exceptions as log-and-continue? (Same concern applies to the similar `except Exception` block in `_create_dag_runs_asset_triggered`.)
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2894619410
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py:
##########
@@ -494,6 +495,11 @@ def trigger_dag_run(
current_user_id = user.get_id()
dag_run.note = (dag_run_note, current_user_id)
return dag_run
+ except BundleVersionUnavailable as e:
+ raise HTTPException(
+ status.HTTP_503_SERVICE_UNAVAILABLE,
+ f"Bundle version not yet available. Please retry: {e}",
Review Comment:
FastAPI's HTTPException already returns structured JSON (`{"detail": "..."}`). This matches the existing patterns in the codebase.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2896001717
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,46 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
)
- )
- dag_run.consumed_asset_events.extend(asset_events)
- session.flush()
- apdr.created_dag_run_id = dag_run.id
- session.flush()
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except Exception:
Review Comment:
`OperationalError` is still narrower than the retry contract for `@retry_db_transaction`, which is based on `DBAPIError`. Other `DBAPIError` subclasses (e.g. `IntegrityError`) can still be caught by the broad `except Exception` and get swallowed. Could we switch this to `except DBAPIError: raise` (or rollback + re-raise), then keep non-DB exceptions as log-and-continue? Same applies to the analogous block in `_create_dag_runs_asset_triggered`.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1970,30 +1970,44 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
partition_dag_ids.add(apdr.target_dag_id)
run_after = timezone.utcnow()
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
- ),
- logical_date=None,
- data_interval=None,
- partition_key=apdr.partition_key,
- run_after=run_after,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- asset_events = session.scalars(
- select(AssetEvent).where(
- PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
- PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after
+ ),
+ logical_date=None,
+ data_interval=None,
+ partition_key=apdr.partition_key,
+ run_after=run_after,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
+ )
+ asset_events = session.scalars(
+ select(AssetEvent).where(
+ PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+ PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+ )
+ )
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.flush()
+ apdr.created_dag_run_id = dag_run.id
+ session.flush()
+ except BundleVersionUnavailable:
+ self.log.warning(
+ "Bundle version not yet available for partitioned asset-triggered DAG %s. "
+ "Run will be created on next evaluation cycle.",
+ apdr.target_dag_id,
+ )
+ except OperationalError:
Review Comment:
Thanks for the quick follow-up here. One remaining concern: re-raising only `OperationalError` is still narrower than the retry contract for `@retry_db_transaction`, which is based on `DBAPIError`. Other `DBAPIError` subclasses (e.g. `IntegrityError`) can still be caught by the broad `except Exception` and get swallowed. Could we switch this to `except DBAPIError: raise` (or rollback + re-raise), then keep non-DB exceptions as log-and-continue? Same applies to the analogous block in `_create_dag_runs_asset_triggered`.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4004342067
> I still don't see the value of this. Running a dagrun on the latest or old version(the version the dagrun was created with initially) should be a dagrun clearing behaviour. Are you saying that if a Dag is created(when parsed) and it has bundle version 1, then after say 3 runs, and changing the bundle version 3 times, that when we set "run_on_latest_version=False", then every time a new dagrun is created, the first bundle version would be used? Is there a business case for this?
No, it doesn't pin to the first-ever version. `DagModel.bundle_version` is updated every time the DAG is re-parsed (see `dag_processing/collection.py:608`), so it always reflects the version the DAG was most recently parsed from.
The distinction is between two values:
- **DagModel.bundle_version**: the version the DAG was last parsed from
- **DagBundleModel.version**: the latest version the bundle knows about (e.g. the latest git commit)
These can differ temporarily. For example, a Git bundle detects a new commit and updates `DagBundleModel.version`, but the scheduler hasn't re-parsed the DAG files from that new commit yet. With `run_on_latest_version=False`, the dagrun is created with the parsed version (which the scheduler has validated). With True, it uses the latest bundle version even if parsing hasn't caught up.
But the primary motivation here is giving people control over the default behavior when rerunning DAGs and tasks. When a user clears a dagrun that was created with a specific bundle version, the UI shows a checkbox asking whether to rerun with the latest version or the version the original run used. The `run_on_latest_version` config controls the default state of that checkbox at the global and DAG level, so teams don't have to make that decision manually every time.
The business case is reproducibility. If you're debugging a failed run and clear it, False means the rerun uses the same code as the original run. True means it picks up whatever the latest version is, which could include unrelated changes that make it harder to isolate the issue. Different teams will have different preferences here, and this gives them a way to set that default without thinking about it on every clear/rerun.
I agree the docs need to be clearer about this. I'll rewrite the section to explain the distinction between "parsed version" and "latest version" properly and make the rerun use case more prominent.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4038658152
From my perspective, I would favour Option C for now. Keep `run_on_latest_version` as a separate concern (it controls rerun behaviour, not whether versions are tracked), document the distinction clearly, and propose consolidation as a separate issue where the design can be discussed properly without holding up this feature.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4033293868
I'm still uncomfortable with the overlap this has with the existing `disable_bundle_versioning` flag/config.
History wise: I think it is reasonable to want to track bundle_version history. But in practice it doesn't actually help since bundle_version is on dag_run and gets overwritten on clear anyway, right?
I can see a use case for having both an option to force the latest behavior, while also allowing a default and letting the user have the final say. I just wonder if we can consolidate that into a single flag/config instead? These are so closely related...
At the very least, we must explain in docs why they both exist and how they differ. It is not immediately obvious, especially to the end users who'd be making these choices.
I also think we should use `DagModel.bundle_verison` (like we do already elsewhere) vs `DagBundleModel.version`. It could be many many minutes before the dag is reparsed, and that's a pretty long time to have a broken dagrun blowing `BundleVersionUnavailable`s imo.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-4037395009
@jedcunningham Thanks fore the feedback.
We have come from a situation prior to Airflow 3 where the only option was to always rerun with latest DAG code to a situation now where the default position when enabled is that reruns occur with the code as it was at the time. This is already create confusion amongst people I work with day-to-day because it's easy to forget to check a box when rerunning.
In my opinion, this default behaviour has to be something that is configurable at different levels and that also propagates across different DAGs through triggers (there's another issue I have waiting to address after this).
If we can agree on the principle at least, I'm more than happy to work towards a solution that satisfies and if you are able to provide any more guidance I'd be greatly appreciative.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2894616262
##########
airflow-core/src/airflow/ui/src/components/Clear/useRunOnLatestVersion.ts:
##########
@@ -0,0 +1,94 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useState } from "react";
+
+import { useConfig } from "src/queries/useConfig";
+
+type UseRunOnLatestVersionProps = {
+ /**
+ * Current DAG bundle version
+ */
+ currentBundleVersion?: string | null;
+
+ /**
+ * DAG-level configuration from the DAG details.
+ * If defined (true or false), takes precedence over global config.
+ */
+ dagLevelConfig?: boolean | null;
+
+ /**
+ * DAG run bundle version
+ */
+ runBundleVersion?: string | null;
+};
+
+type UseRunOnLatestVersionResult = {
+ /**
+ * Setter for the checkbox value
+ */
+ setValue: (value: boolean) => void;
+
+ /**
+ * Whether the checkbox should be shown
+ */
+ shouldShowCheckbox: boolean;
+
+ /**
+ * Current value of the checkbox (controlled)
+ */
+ value: boolean;
+};
+
+/**
+ * Custom hook for managing "Run on Latest Version" checkbox state.
+ *
+ * Implements the three-level precedence hierarchy:
+ * 1. DAG-level configuration (highest priority)
+ * 2. Global configuration
+ * 3. System default (false)
+ *
+ * Uses nullable override pattern: state is null until user interacts,
+ * avoiding the useState + useEffect synchronization anti-pattern.
+ */
+export const useRunOnLatestVersion = ({
+ currentBundleVersion,
+ dagLevelConfig,
+ runBundleVersion,
+}: UseRunOnLatestVersionProps): UseRunOnLatestVersionResult => {
+ const globalConfigValue = useConfig("run_on_latest_version");
+ const globalDefault = Boolean(globalConfigValue);
+
+ // Precedence: DAG-level > Global > System default (false)
+ const defaultValue = dagLevelConfig ?? globalDefault;
+
+ // Nullable override: null until user interacts, then stores their choice
+ const [userOverride, setUserOverride] = useState<boolean | null>(null);
+ const value = userOverride ?? defaultValue;
+
+ // Show checkbox only when versions differ and run has a valid bundle version
+ const hasValidRunVersion =
+ runBundleVersion !== undefined && runBundleVersion !== null && runBundleVersion !== "";
+ const shouldShowCheckbox = hasValidRunVersion && currentBundleVersion !== runBundleVersion;
Review Comment:
The prop is typed as `string | null | undefined` via the optional `?`. The comparison logic handles all cases correctly so, when undefined, the checkbox simply won't show.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871547730
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2073,22 +2084,40 @@ def _create_dag_runs_asset_triggered(
)
)
- dag_run = dag.create_dagrun(
- run_id=DagRun.generate_run_id(
- run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
- ),
- logical_date=None,
- data_interval=None,
- run_after=triggered_date,
- run_type=DagRunType.ASSET_TRIGGERED,
- triggered_by=DagRunTriggeredByType.ASSET,
- state=DagRunState.QUEUED,
- creating_job_id=self.job.id,
- session=session,
- )
- Stats.incr("asset.triggered_dagruns")
- dag_run.consumed_asset_events.extend(asset_events)
- session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))
+ try:
+ dag_run = dag.create_dagrun(
+ run_id=DagRun.generate_run_id(
+ run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date
+ ),
+ logical_date=None,
+ data_interval=None,
+ run_after=triggered_date,
+ run_type=DagRunType.ASSET_TRIGGERED,
+ triggered_by=DagRunTriggeredByType.ASSET,
+ state=DagRunState.QUEUED,
+ creating_job_id=self.job.id,
+ session=session,
+ )
+ Stats.incr("asset.triggered_dagruns")
+ dag_run.consumed_asset_events.extend(asset_events)
+ session.execute(
+ delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id)
+ )
+ except BundleVersionUnavailable:
Review Comment:
You're right! This was the one create_dagrun call site without BundleVersionUnavailable handling. Wrapped it in the same try/except pattern used in `_create_dag_runs` and `_create_dag_runs_asset_triggered`. When the exception fires, `apdr.created_dag_run_id` remains `None` so it's retried on the next evaluation cycle. Added a test for this path as well.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on code in PR #61448:
URL: https://github.com/apache/airflow/pull/61448#discussion_r2871591146
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -1113,13 +1225,27 @@ def _create_orm_dagrun(
partition_key: str | None = None,
session: Session = NEW_SESSION,
) -> DagRun:
- bundle_version = None
- 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)
+ bundle_version = _resolve_bundle_version(
+ dag=dag,
+ session=session,
+ )
+ dag_version = DagVersion.get_latest_version(dag.dag_id, bundle_version=bundle_version, session=session)
+
if not dag_version:
+ if bundle_version:
+ # Bundle version exists but not yet serialized - this is a temporary race condition
+ log.warning(
+ "Bundle version %s for DAG %s is not yet available. "
Review Comment:
I haven't reproduced it in a live system but the race condition was identified through code analysis. It's a pretty narrow window in practice and it's more likely to surface if the DAG files are large or slow to parse, or the scheduler is under heavy load.
I validated the scenario with a unit test (`test_create_dagrun_race_condition_bundle_version_not_serialized_yet`) that sets up exactly this state, DagBundleModel at v2.0.0 but only a DagVersion for v1.0.0, and confirms BundleVersionUnavailable is raised with a clear retry message.
--
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] Add three-level `run_on_latest_version` configuration hierarchy [airflow]
Posted by "nathadfield (via GitHub)" <gi...@apache.org>.
nathadfield commented on PR #61448:
URL: https://github.com/apache/airflow/pull/61448#issuecomment-3858842712
> My immediate reaction is, not sure hows its much different than the existing [[dag_processor] disable_bundle_versioning](https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#disable-bundle-versioning) and `disable_bundle_versioning` on Dag. My 30s review for the day :)
The key distinction is that these features serve different purposes:
disable_bundle_versioning is binary - versioning on or off:
- It's a structural decision about whether to use bundle versioning at all
- No middle ground
run_on_latest_version adds flexibility at the DAG/task level when rerunning:
- Configures what happens when you clear, rerun, or trigger a DAG
- Do you want to rerun with the latest code, or stick with the original version?
Without this feature, when using bundle versioning you have a binary choice:
- Turn off versioning → reruns get latest code (but lose version tracking)
- Keep versioning → stuck manually selecting versions in the UI for every rerun
--
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