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