From 80a6f75b0010e1852f169cee1a750069d6e836bf Mon Sep 17 00:00:00 2001 From: ColtenOuO Date: Sun, 23 Aug 2026 10:12:22 +0000 Subject: [PATCH 1/2] Fix manually triggered runs blocking a scheduled run at the same logical_date A manually triggered Dag run's logical_date can coincide with a later, unrelated scheduled run's logical_date -- they represent different data intervals despite sharing that value, since a manual trigger's data interval is derived from when it fires rather than from the period it is labeled with. The DagRun table's uniqueness constraint didn't distinguish between run types, so whichever run landed first silently and permanently claimed that logical_date, and the scheduler treated the other as a duplicate it had "already handled" -- dropping that scheduling period for good, with only an easily-missed WARNING log as a trace. Verified directly against the metadata database that this isn't just an in-memory dedup-check problem: a manual run occupying a logical_date makes the scheduled run's INSERT fail on the same unique-constraint violation regardless of how the scheduler checks for duplicates beforehand, so the fix widens what the database itself considers a duplicate rather than just how the scheduler looks for one. --- airflow-core/docs/migrations-ref.rst | 5 +- .../src/airflow/jobs/scheduler_job_runner.py | 3 +- ..._dag_run_logical_date_unique_constraint.py | 88 +++++++++++++++++++ airflow-core/src/airflow/models/backfill.py | 7 +- airflow-core/src/airflow/models/dagrun.py | 4 +- airflow-core/src/airflow/utils/db.py | 2 +- .../tests/unit/jobs/test_scheduler_job.py | 40 ++++++++- .../tests/unit/models/test_backfill.py | 57 ++++++++++++ 8 files changed, 196 insertions(+), 10 deletions(-) create mode 100644 airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_run_type_to_dag_run_logical_date_unique_constraint.py diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index 8e03f37ef5430..8c312a301f51a 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/airflow-core/docs/migrations-ref.rst @@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | Revision ID | Revises ID | Airflow Version | Description | +=========================+==================+===================+==============================================================+ -| ``8d3f1a6b2c47`` (head) | ``c7f0a5d2e9b4`` | ``3.4.0`` | Add team_name to log. | +| ``60e5d30563ae`` (head) | ``8d3f1a6b2c47`` | ``3.4.0`` | Add run_type to the DagRun (dag_id, logical_date) unique | +| | | | constraint. | ++-------------------------+------------------+-------------------+--------------------------------------------------------------+ +| ``8d3f1a6b2c47`` | ``c7f0a5d2e9b4`` | ``3.4.0`` | Add team_name to log. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | ``c7f0a5d2e9b4`` | ``76c46545c91e`` | ``3.4.0`` | Lower case team names. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index f8b2df4aac2ca..325551e960c11 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -2533,7 +2533,8 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) - .where( tuple_(DagRun.dag_id, DagRun.logical_date).in_( (dm.dag_id, dm.next_dagrun) for dm in dag_models - ) + ), + DagRun.run_type == DagRunType.SCHEDULED, ) .options(load_only(DagRun.dag_id, DagRun.logical_date)) ).all() diff --git a/airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_run_type_to_dag_run_logical_date_unique_constraint.py b/airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_run_type_to_dag_run_logical_date_unique_constraint.py new file mode 100644 index 0000000000000..f448d0a0866de --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_run_type_to_dag_run_logical_date_unique_constraint.py @@ -0,0 +1,88 @@ +# +# 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. + +""" +Add run_type to the DagRun (dag_id, logical_date) unique constraint. + +Revision ID: 60e5d30563ae +Revises: 8d3f1a6b2c47 +Create Date: 2026-08-22 21:15:00.000000 + +""" + +from __future__ import annotations + +from alembic import context, op +from sqlalchemy import text + +# revision identifiers, used by Alembic. +revision = "60e5d30563ae" +down_revision = "8d3f1a6b2c47" +branch_labels = None +depends_on = None +airflow_version = "3.4.0" + +_OLD_CONSTRAINT = "dag_run_dag_id_logical_date_key" +_NEW_CONSTRAINT = "dag_run_dag_id_logical_date_run_type_key" + +_DUPLICATE_LOGICAL_DATE_QUERY = ( + "select dr.* from dag_run dr " + "where (dr.dag_id, dr.logical_date) in (" + " select dag_id, logical_date from dag_run group by dag_id, logical_date having count(*) > 1" + ") " + "and dr.id not in (" + " select min(id) from dag_run group by dag_id, logical_date having count(*) > 1" + ")" +) + + +def upgrade(): + with op.batch_alter_table("dag_run", schema=None) as batch_op: + batch_op.drop_constraint(_OLD_CONSTRAINT, type_="unique") + batch_op.create_unique_constraint(_NEW_CONSTRAINT, columns=["dag_id", "logical_date", "run_type"]) + + +def _move_offending_dagruns(): + from airflow.utils.db import AIRFLOW_MOVED_TABLE_PREFIX + + conn = op.get_bind() + offline = context.is_offline_mode() + + if not offline and conn.execute(text(_DUPLICATE_LOGICAL_DATE_QUERY)).fetchone() is None: + return + + offending_table_name = f"{AIRFLOW_MOVED_TABLE_PREFIX}__3_4_0__offending_dag_run_logical_date" + if conn.dialect.name == "mysql": + op.execute(f"create table {offending_table_name} like dag_run") + op.execute(f"insert into {offending_table_name} {_DUPLICATE_LOGICAL_DATE_QUERY}") + else: + op.execute(f"create table {offending_table_name} as {_DUPLICATE_LOGICAL_DATE_QUERY}") + + if offline: + op.execute(f"-- TODO: DAG runs unable to be downgraded are moved to {offending_table_name}.") + op.execute(f"-- TODO: Table {offending_table_name} can be removed after contained data are reviewed.") + + op.execute(f"delete from dag_run where id in (select id from {offending_table_name})") + + +def downgrade(): + _move_offending_dagruns() + + with op.batch_alter_table("dag_run", schema=None) as batch_op: + batch_op.drop_constraint(_NEW_CONSTRAINT, type_="unique") + batch_op.create_unique_constraint(_OLD_CONSTRAINT, columns=["dag_id", "logical_date"]) diff --git a/airflow-core/src/airflow/models/backfill.py b/airflow-core/src/airflow/models/backfill.py index 13b9e6d004f6a..f6cc1e5cc77be 100644 --- a/airflow-core/src/airflow/models/backfill.py +++ b/airflow-core/src/airflow/models/backfill.py @@ -405,10 +405,7 @@ def _create_backfill_dag_run_non_partitioned( return lock = session.execute( with_row_locks( - query=select(DagRun).where( - DagRun.logical_date == info.logical_date, - DagRun.dag_id == dag.dag_id, - ), + query=select(DagRun).where(DagRun.id == dr.id), session=session, skip_locked=True, ) @@ -599,7 +596,7 @@ def _handle_clear_run( # Update backfill_id, run_type, and optionally conf in DagRun table stmt = ( update(DagRun) - .where(DagRun.logical_date == info.logical_date, DagRun.dag_id == dag.dag_id) + .where(DagRun.id == dr.id) .values( backfill_id=backfill_id, run_type=DagRunType.BACKFILL_JOB, diff --git a/airflow-core/src/airflow/models/dagrun.py b/airflow-core/src/airflow/models/dagrun.py index 46fbebb080038..f9f5abc1bfe0c 100644 --- a/airflow-core/src/airflow/models/dagrun.py +++ b/airflow-core/src/airflow/models/dagrun.py @@ -329,7 +329,9 @@ class DagRun(Base, LoggingMixin): __table_args__ = ( Index("dag_id_state", dag_id, _state), UniqueConstraint("dag_id", "run_id", name="dag_run_dag_id_run_id_key"), - UniqueConstraint("dag_id", "logical_date", name="dag_run_dag_id_logical_date_key"), + UniqueConstraint( + "dag_id", "logical_date", "run_type", name="dag_run_dag_id_logical_date_run_type_key" + ), Index("idx_dag_run_dag_id", dag_id), Index("idx_dag_run_run_after", run_after), Index("idx_dag_run_created_dag_version_id", created_dag_version_id), diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 71ec2d2004b6d..87edd92b16b22 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -117,7 +117,7 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "8d3f1a6b2c47", + "3.4.0": "60e5d30563ae", } # Prefix used to identify tables holding data moved during migration. diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index a7b332e763f1c..36369773f5fb5 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -6686,6 +6686,44 @@ def test_scheduler_create_dag_runs_check_existing_run(self, dag_maker, session): assert dag_model.next_dagrun == DEFAULT_DATE + timedelta(days=1) session.rollback() + def test_scheduler_create_dag_runs_manual_run_does_not_block_scheduled_run(self, dag_maker, session): + with dag_maker( + dag_id="test_scheduler_create_dag_runs_manual_run_does_not_block_scheduled_run", + schedule=timedelta(days=1), + catchup=True, + ) as dag: + EmptyOperator(task_id="dummy") + + dag_model = dag_maker.dag_model + assert dag_model.next_dagrun == DEFAULT_DATE + + manual_run = dag_maker.create_dagrun( + run_type=DagRunType.MANUAL, + logical_date=dag_model.next_dagrun, + start_date=timezone.utcnow(), + state=State.RUNNING, + session=session, + creating_job_id=2, + ) + session.flush() + + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + + self.job_runner._create_dag_runs([dag_model], session) + + scheduled_run = session.scalars( + select(DagRun).where( + DagRun.dag_id == dag.dag_id, + DagRun.run_type == DagRunType.SCHEDULED, + DagRun.logical_date == DEFAULT_DATE, + ) + ).one_or_none() + assert scheduled_run is not None, "the scheduled run must still be created despite the manual run" + assert scheduled_run.id != manual_run.id + assert dag_model.next_dagrun == DEFAULT_DATE + timedelta(days=1) + session.rollback() + @conf_vars({("scheduler", "use_job_schedule"): "false"}) def test_do_schedule_max_active_runs_dag_timed_out(self, dag_maker, session): """Test that tasks are set to a finished state when their DAG times out""" @@ -6920,7 +6958,7 @@ def test_do_schedule_max_active_runs_and_manual_trigger(self, dag_maker, mock_ex BashOperator(task_id="dummy3", bash_command="true") session = settings.Session() - dag_run = dag_maker.create_dagrun(state=State.QUEUED, session=session) + dag_run = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED, state=State.QUEUED, session=session) scheduler_job = Job() self.job_runner = SchedulerJobRunner(job=scheduler_job) diff --git a/airflow-core/tests/unit/models/test_backfill.py b/airflow-core/tests/unit/models/test_backfill.py index ea621be84d6c1..57c3f86a9b040 100644 --- a/airflow-core/tests/unit/models/test_backfill.py +++ b/airflow-core/tests/unit/models/test_backfill.py @@ -1527,3 +1527,60 @@ def test_handle_clear_run_preserves_partition_key(dag_maker, session): ) assert bdr is not None assert bdr.partition_key == partition_key + + +def test_handle_clear_run_only_updates_the_targeted_run(dag_maker, session): + with dag_maker(schedule="@daily") as dag: + PythonOperator(task_id="hi", python_callable=print) + + logical_date = timezone.parse("2026-01-10") + dr = dag_maker.create_dagrun( + run_id="scheduled_2026-01-10", + run_type=DagRunType.SCHEDULED, + logical_date=logical_date, + session=session, + state="failed", + ) + other_manual_run = dag_maker.create_dagrun( + run_id="manual_2026-01-10", + run_type=DagRunType.MANUAL, + logical_date=logical_date, + session=session, + state="success", + ) + session.commit() + + backfill = Backfill( + dag_id=dag.dag_id, + from_date=logical_date, + to_date=logical_date, + dag_run_conf=None, + max_active_runs=1, + reprocess_behavior=ReprocessBehavior.FAILED, + ) + session.add(backfill) + session.flush() + + info = DagRunInfo( + run_after=logical_date, + data_interval=DataInterval(logical_date, logical_date), + partition_key=None, + partition_date=None, + ) + + _handle_clear_run( + session=session, + dag=dag, + dr=dr, + info=info, + backfill_id=backfill.id, + sort_ordinal=1, + ) + session.flush() + session.refresh(dr) + session.refresh(other_manual_run) + + assert dr.run_type == DagRunType.BACKFILL_JOB + assert dr.backfill_id == backfill.id + assert other_manual_run.run_type == DagRunType.MANUAL + assert other_manual_run.backfill_id is None From e6a5e870ca6e0131320ce1818485d0f6444f7686 Mon Sep 17 00:00:00 2001 From: ColtenOuO Date: Sun, 23 Aug 2026 10:45:56 +0000 Subject: [PATCH 2/2] Add newsfragment --- airflow-core/newsfragments/72003.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/72003.bugfix.rst diff --git a/airflow-core/newsfragments/72003.bugfix.rst b/airflow-core/newsfragments/72003.bugfix.rst new file mode 100644 index 0000000000000..3fc7d17918188 --- /dev/null +++ b/airflow-core/newsfragments/72003.bugfix.rst @@ -0,0 +1 @@ +Fixed a bug where a manually triggered Dag run whose ``logical_date`` happened to match a later scheduled run's ``logical_date`` would silently and permanently block that scheduled run from ever being created. The two runs represent different data intervals despite sharing a ``logical_date``, and are now distinguished by ``run_type`` in the metadata database.