hanxdatadog opened a new issue, #76041: URL: https://github.com/apache/airflow/issues/76041
### Is your feature request related to a problem? Bundle versioning lets runs execute the code of a chosen bundle version, and #60484 (implemented by #61550 in 3.4) lets a run be pinned to a version that already has a `DagVersion`. There is still no supported way to run a Dag at a bundle version the Dag processor never parsed, e.g. an unmerged commit, while keeping that commit's **task graph and params**. Today such a run would execute the commit's task code against the currently deployed graph, so added or removed tasks, changed dependencies, retries or params don't take effect. The run should stay a run of the same `dag_id`, so XComs, pools and other `dag_id`-keyed behavior are unchanged. The natural implementation is to parse the Dag at that bundle version and store an extra `DagVersion` + `SerializedDagModel` row that the run is pinned to by id. That row must never be treated as the Dag's current version. But Airflow has no single definition of "current version". Most code takes the highest `version_number` per `dag_id`, while some paths take the newest row by time (references at `3.4.0rc1`): | Code path | Picks by | Effect if a newer, non-latest row exists | |---|---|---| | [`db_cleanup` `dag_version` config](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/utils/db_cleanup.py#L315-L328) / [`_subquery_keep_last`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/utils/db_cleanup.py#L646-L662) | `max(created_at)` | `airflow db clean` can delete the real latest version | | [`SerializedDagModel.write_dag` deadline reuse](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/serialized_dag.py#L664-L668) | `created_at` desc | Deadline ids aren't reused, so the hash changes and the latest version is rewritten on every parse | | [`SerializedDagModel.get_dag_dependencies`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/serialized_dag.py#L1007-L1018) | `max(created_at)` | Wrong Dag dependency graph in the UI/CLI | | [`DagRun.dag_versions`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/dagrun.py#L556-L578) | sort by `id` | Wrong "current" version in the grid and OpenLineage | | [`DagCode` latest](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/dagcode.py#L182-L189) | `last_updated` desc | Wrong source shown | | [`DBDagBag.iter_all_latest_version_dags`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/dagbag.py#L219-L230) | no latest filter | Returns every version (currently unused) | Run creation and scheduling ([`_get_current_dag`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L401-L407), [`_create_orm_dagrun`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/serialization/definitions/dag.py#L1460-L1469), [`DBDagBag._version_from_dag_run`](https://github.com/apache/airflow/blob/3.4.0rc1/airflow-core/src/airflow/models/dagbag.py#L196-L212), clear on latest) use `version_number`, so they are safe today. That safety comes only from each query happening to order by number. As a result, every workaround is fragile: - storing the row under a synthetic `dag_id` (invisible to the UI); - renumbering versions so the extra row is never the highest, and backdating timestamps; - patching each reader. ### Proposed solution 1. **One "latest version" rule.** Add a single helper/predicate (e.g. on `DagVersion`) for "the Dag's current version", and make every reader above use it instead of its own ordering. This is useful on its own: the time-ordered readers can already disagree with the scheduler (e.g. after in-place refreshes, #73861). 2. **First-class non-latest versions.** Let a `DagVersion` be marked as not eligible to become current (for example a column, or a version kind), set when it's created for a specific run rather than by the Dag processor's normal parse. The helper from (1) excludes it; the UI still lists it and shows its graph for runs pinned to it. 3. **Trigger at an unparsed bundle version.** Extend the trigger-at-bundle-version path from #60484: when no `DagVersion` exists for the requested bundle version, parse the Dag at that version (Dag processor or a parse-on-demand endpoint), store it as a non-latest version, validate the run's conf against its params, and pin the run to it. ### Alternative solutions you considered - **A separate Dag (`adhoc_<dag_id>`)**: changes `dag_id`, so XComs, pools and dependencies differ from the real Dag. - **A synthetic hidden Dag row holding the version**: works with no core changes, but the UI shows the deployed graph for the run. - **Low or negative `version_number`s for the extra rows**: safe for scheduling, but every time-ordered reader above still picks them, and the version labels are confusing. - **Re-inserting a copy of the deployed version on top in the same transaction**: races with the Dag processor's in-place updates. ### Who would benefit? DAG authors who want to test an unmerged change of a Dag's structure or params against the real Dag in a shared environment, without deploying it, and operators running bundle-versioned deployments. ### Additional context Related: #60484 (trigger at an existing bundle version), #73861 (in-place `bundle_version` refresh), #71453 (`created_dag_version_id` contract). Happy to contribute (1) as a first PR. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
