This is an automated email from the ASF dual-hosted git repository. kaxilnaik pushed a commit to branch v3-0-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 364d2468a18defafadeadffe18e73b8848a6112c Author: Kaxil Naik <[email protected]> AuthorDate: Mon Aug 11 18:25:51 2025 +0100 Fix `DetachedInstanceError` when accessing `DagRun.created_dag_version` (#54362) The error occurred when trying to access `dag_run.created_dag_version` on a DagRun object that was detached from its SQLAlchemy session. Stacktrace: ``` [2025-08-08T19:03:28.761+0000] {scheduler_job_runner.py:935} ERROR - Marking task instance <TaskInstance: sell_oms_shipconfirm_mao_to_ecom_load_delta_quarterly.shipconfirm_delta_to_kafka_notebook_task scheduled__2025-08-08T07:30:00+00:00 [running]> as failed Traceback (most recent call last): File "/usr/local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 925, in process_executor_events dag = scheduler_dag_bag.get_dag(dag_run=ti.dag_run, session=session) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 141, in get_dag version = self._version_from_dag_run(dag_run=dag_run, session=session) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 135, in _version_from_dag_run dag_version = dag_run.created_dag_version ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/attributes.py", line 487, in __get__ return self.impl.get(state, dict_) ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/attributes.py", line 959, in get value = self._fire_loader_callables(state, key, passive) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/attributes.py", line 995, in _fire_loader_callables return self.callable_(state, passive) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/sqlalchemy/orm/strategies.py", line 863, in _load_for_state raise orm_exc.DetachedInstanceError( sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <DagRun at 0x7f46d309a5d0> is not bound to a Session; lazy load operation of attribute 'created_dag_version' cannot proceed (Background on this error at: https://sqlalche.me/e/14/bhk3) ``` Two-layer fix: 1. Proactive loading: Add `selectinload` for `created_dag_version` in `process_executor_events` query to ensure optimal performance for this critical path 2. Defensive code: Add relationship loading check in `DBDagBag._version_from_dag_run` to handle all call sites safely using SQLAlchemy inspect pattern (cherry picked from commit 05d2229d0e1b3aeec7b196ef007384d16350f3a1) --- .../src/airflow/jobs/scheduler_job_runner.py | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 6e843526831..c46f99d703d 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -32,9 +32,10 @@ from functools import lru_cache, partial from itertools import groupby from typing import TYPE_CHECKING, Any, Callable -from sqlalchemy import and_, delete, desc, exists, func, select, text, tuple_, update +from sqlalchemy import and_, delete, desc, exists, func, inspect, select, text, tuple_, update from sqlalchemy.exc import OperationalError from sqlalchemy.orm import joinedload, lazyload, load_only, make_transient, selectinload +from sqlalchemy.orm.attributes import NO_VALUE from sqlalchemy.sql import expression from airflow import settings @@ -131,11 +132,19 @@ class SchedulerDagBag: @staticmethod def _version_from_dag_run(dag_run, session): - if dag_run.bundle_version: - dag_version = dag_run.created_dag_version - else: + if not dag_run.bundle_version: dag_version = DagVersion.get_latest_version(dag_id=dag_run.dag_id, session=session) - return dag_version + if dag_version: + return dag_version + + # Check if created_dag_version relationship is already loaded to avoid DetachedInstanceError + info = inspect(dag_run) + if info.attrs.created_dag_version.loaded_value is not NO_VALUE: + # Relationship is already loaded, safe to access + return dag_run.created_dag_version + + # Relationship not loaded, fetch it explicitly from current session + return session.get(DagVersion, dag_run.created_dag_version_id) def get_dag(self, dag_run: DagRun, session: Session) -> DAG | None: version = self._version_from_dag_run(dag_run=dag_run, session=session) @@ -827,7 +836,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin): select(TI) .where(filter_for_tis) .options(selectinload(TI.dag_model)) - .options(selectinload(TI.dag_run).selectinload(DagRun.consumed_asset_events)) + .options(joinedload(TI.dag_run).selectinload(DagRun.consumed_asset_events)) + .options(joinedload(TI.dag_run).selectinload(DagRun.created_dag_version)) .options(joinedload(TI.dag_version)) ) # row lock this entire set of taskinstances to make sure the scheduler doesn't fail when we have
