Fix manually triggered runs blocking a scheduled run at the same logical_date

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.                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

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.

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()

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"])

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,

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),

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.

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)

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