This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 604bcece5be [v3-4-test] Fix dynamically mapped tasks never running on
scheduled Dag runs (#74861) (#74918)
604bcece5be is described below
commit 604bcece5bec7e80ed5e26a1d01484dd8f2d843e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Oct 18 11:28:59 2026 +0530
[v3-4-test] Fix dynamically mapped tasks never running on scheduled Dag
runs (#74861) (#74918)
DynamicRegion.get_or_create_many() uses a SAVEPOINT (session.begin_nested())
to safely insert dynamic-mapping regions. Releasing that SAVEPOINT
dispatches
the same before_commit event as a real commit, and the scheduler's
CommitProhibitorGuard rejected it unconditionally, raising "UNEXPECTED
COMMIT - THIS WILL BREAK HA LOCKS!" before the Dag run's task instances
could be created.
This only happened for a Dag run the scheduler created from its own
schedule: the API server's Dag run creation runs outside the scheduler's
guarded session, so the identical SAVEPOINT release there was harmless. A
dynamically mapped task on a real (non-manually-triggered) schedule could
therefore never run, with the same scheduling error repeating every loop.
Releasing a savepoint ends neither the outer transaction nor its row locks,
so it cannot break the HA locks this guard protects -- only the outer
commit can, and that remains guarded.
(cherry picked from commit f5038e678343787f676e1bb6f0923c06f4b4f4b3)
Co-authored-by: Rahul Vats <[email protected]>
---
airflow-core/src/airflow/utils/sqlalchemy.py | 9 ++++++++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 21 +++++++++++++++++++++
airflow-core/tests/unit/utils/test_sqlalchemy.py | 20 ++++++++++++++++++++
3 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py
b/airflow-core/src/airflow/utils/sqlalchemy.py
index 6b7db82b8ef..2d72ef319a8 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -690,7 +690,14 @@ class CommitProhibitorGuard:
def __init__(self, session: Session):
self.session = session
- def _validate_commit(self, _):
+ def _validate_commit(self, session):
+ # before_commit fires for a SAVEPOINT release too, not just
session.commit(). A
+ # savepoint release ends neither the outer transaction nor its row
locks, so it can't
+ # break the HA locks this guard protects -- skip the check while one
is open. This
+ # covers session.commit() and savepoint exits; nothing under this
guard commits a
+ # transaction handle directly today, which would bypass this check.
+ if session.in_nested_transaction():
+ return
if self.expected_commit:
self.expected_commit = False
return
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 858ffc87c4c..576cb59a898 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -6585,6 +6585,27 @@ class TestSchedulerJob:
assert dr.start_date is None
assert dr.creating_job_id == scheduler_job.id
+ def
test_do_scheduling_creates_task_instances_for_scheduler_created_mapped_run(self,
dag_maker, session):
+ """A Dag run the scheduler creates itself (not via
dag_maker.create_dagrun()) for a
+ dynamically mapped task must get its mapped TaskInstances created.
+
+ Only reproducible through the real CommitProhibitorGuard installed by
_do_scheduling:
+ other tests in this file exercise
_create_dagruns_for_dags/_schedule_all_dag_runs
+ directly with the plain session cast as the guard
(`cast("CommitProhibitorGuard",
+ session)`), which never installs the real before_commit listener.
+ """
+ with dag_maker(dag_id="test_scheduled_mapped_task", schedule="@once",
session=session):
+ BashOperator.partial(task_id="mapped",
bash_command="true").expand(env=[{"n": "1"}, {"n": "2"}])
+
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job,
executors=[self.null_exec])
+
+ self.job_runner._do_scheduling(session)
+
+ dr = session.scalars(select(DagRun).where(DagRun.dag_id ==
"test_scheduled_mapped_task")).one()
+ tis = dr.get_task_instances(session=session)
+ assert sorted(ti.map_index for ti in tis) == [0, 1]
+
@pytest.mark.need_serialized_dag
@conf_vars({("scheduler", "batch_asset_events"): "True"})
def test_create_dag_runs_assets(self, session, dag_maker):
diff --git a/airflow-core/tests/unit/utils/test_sqlalchemy.py
b/airflow-core/tests/unit/utils/test_sqlalchemy.py
index e5e883eac85..5cb11e8cb41 100644
--- a/airflow-core/tests/unit/utils/test_sqlalchemy.py
+++ b/airflow-core/tests/unit/utils/test_sqlalchemy.py
@@ -219,6 +219,26 @@ class TestSqlAlchemyUtils:
other_session.execute(text("SELECT 1"))
other_session.commit()
+ def test_prohibit_commit_allows_savepoint_release(self):
+ """A SAVEPOINT release (session.begin_nested()) dispatches the same
before_commit
+ event as a real commit, but must not trip the guard: it ends neither
the outer
+ transaction nor its row locks."""
+ with prohibit_commit(self.session):
+ self.session.execute(text("SELECT 1"))
+ with self.session.begin_nested():
+ self.session.execute(text("SELECT 1"))
+
+ # The guard must still catch a real commit after the savepoint
released.
+ with pytest.raises(RuntimeError, match="UNEXPECTED COMMIT"):
+ self.session.commit()
+ self.session.rollback()
+
+ # ...and an explicit commit while still inside an open savepoint.
+ with pytest.raises(RuntimeError, match="UNEXPECTED COMMIT"):
+ with self.session.begin_nested():
+ self.session.commit()
+ self.session.rollback()
+
def teardown_method(self):
self.session.close()
settings.engine.dispose()
