Add run_after column to DagRun model

airflow/api/common/trigger_dag.py

@@ -102,6 +102,7 @@ def _trigger_dag(
         run_id=run_id,
         logical_date=logical_date,
         data_interval=data_interval,
+        run_after=data_interval.end,
         conf=run_conf,
         run_type=DagRunType.MANUAL,
         triggered_by=triggered_by,

airflow/api_connexion/endpoints/dag_run_endpoint.py

@@ -354,6 +354,7 @@ def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
                 run_id=run_id,
                 logical_date=logical_date,
                 data_interval=data_interval,
+                run_after=data_interval.end,
                 conf=post_body.get("conf"),
                 run_type=DagRunType.MANUAL,
                 triggered_by=DagRunTriggeredByType.REST_API,

airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -377,6 +377,7 @@ def trigger_dag_run(
             run_id=run_id,
             logical_date=logical_date,
             data_interval=data_interval,
+            run_after=data_interval.end,
             conf=body.conf,
             run_type=DagRunType.MANUAL,
             triggered_by=DagRunTriggeredByType.REST_API,

airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -219,6 +219,7 @@ class DagRun(StrictBaseModel):
     logical_date: UtcDateTime
     data_interval_start: UtcDateTime | None
     data_interval_end: UtcDateTime | None
+    run_after: UtcDateTime
     start_date: UtcDateTime
     end_date: UtcDateTime | None
     run_type: DagRunType

airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -144,6 +144,7 @@ def ti_run(
                 DR.dag_id,
                 DR.data_interval_start,
                 DR.data_interval_end,
+                DR.run_after,
                 DR.start_date,
                 DR.end_date,
                 DR.run_type,

airflow/cli/commands/remote_commands/task_command.py

@@ -171,22 +171,26 @@ def _get_dag_run(
         dag_run_logical_date = pendulum.instance(timezone.utcnow())
 
     if create_if_necessary == "memory":
+        data_interval = dag.timetable.infer_manual_data_interval(run_after=dag_run_logical_date)
         dag_run = DagRun(
             dag_id=dag.dag_id,
             run_id=logical_date_or_run_id,
             run_type=DagRunType.MANUAL,
             external_trigger=True,
             logical_date=dag_run_logical_date,
-            data_interval=dag.timetable.infer_manual_data_interval(run_after=dag_run_logical_date),
+            data_interval=data_interval,
+            run_after=data_interval.end,
             triggered_by=DagRunTriggeredByType.CLI,
             state=DagRunState.RUNNING,
         )
         return dag_run, True
     elif create_if_necessary == "db":
+        data_interval = dag.timetable.infer_manual_data_interval(run_after=dag_run_logical_date)
         dag_run = dag.create_dagrun(
             run_id=_generate_temporary_run_id(),
             logical_date=dag_run_logical_date,
-            data_interval=dag.timetable.infer_manual_data_interval(run_after=dag_run_logical_date),
+            data_interval=data_interval,
+            run_after=data_interval.end,
             run_type=DagRunType.MANUAL,
             triggered_by=DagRunTriggeredByType.CLI,
             dag_version=None,

airflow/jobs/scheduler_job_runner.py

@@ -1284,6 +1284,7 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -
                         ),
                         logical_date=dag_model.next_dagrun,
                         data_interval=data_interval,
+                        run_after=dag_model.next_dagrun_create_after,
                         run_type=DagRunType.SCHEDULED,
                         triggered_by=DagRunTriggeredByType.TIMETABLE,
                         dag_version=latest_dag_version,
@@ -1395,6 +1396,7 @@ def _create_dag_runs_asset_triggered(
                     ),
                     logical_date=logical_date,
                     data_interval=data_interval,
+                    run_after=max(logical_dates.values()),
                     run_type=DagRunType.ASSET_TRIGGERED,
                     triggered_by=DagRunTriggeredByType.ASSET,
                     dag_version=latest_dag_version,

airflow/migrations/versions/0058_3_0_0_add_dagrun_run_after.py

@@ -0,0 +1,62 @@
+#
+# 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 DagRun run_after.
+
+Revision ID: 6a9e7a527a88
+Revises: 33b04e4bfa19
+Create Date: 2025-01-16 10:07:34.940948
+"""
+
+from __future__ import annotations
+
+import alembic.op as op
+import sqlalchemy as sa
+
+from airflow.utils.sqlalchemy import UtcDateTime
+
+# Revision identifiers, used by Alembic.
+revision = "6a9e7a527a88"
+down_revision = "33b04e4bfa19"
+branch_labels = None
+depends_on = None
+airflow_version = "3.0.0"
+
+
+def upgrade():
+    """Add DagRun run_after."""
+    op.add_column("dag_run", sa.Column("run_after", UtcDateTime, nullable=True))
+
+    # run_after must be set to after the data interval ends.
+    # In very old runs where the data interval is not available, we'll just use
+    # the logical date. This is wrong (one interval too early), but good enough
+    # since those super old runs should have their data interval ended a long
+    # time ago anyway, and setting
+    op.execute("update dag_run set run_after = coalesce(data_interval_end, logical_date)")
+
+    with op.batch_alter_table("dag_run") as batch_op:
+        batch_op.alter_column("run_after", existing_type=UtcDateTime, nullable=False)
+        batch_op.create_index("idx_dag_run_run_after", ["run_after"], unique=False)
+
+
+def downgrade():
+    """Remove DagRun run_after."""
+    with op.batch_alter_table("dag_run") as batch_op:
+        batch_op.drop_index("idx_dag_run_run_after")
+        batch_op.drop_column("run_after")

airflow/models/backfill.py

@@ -296,6 +296,7 @@ def _create_backfill_dag_run(
             ),
             logical_date=info.logical_date,
             data_interval=info.data_interval,
+            run_after=info.run_after,
             conf=dag_run_conf,
             run_type=DagRunType.BACKFILL_JOB,
             triggered_by=DagRunTriggeredByType.BACKFILL,

airflow/models/baseoperator.py

@@ -632,6 +632,7 @@ def run(
                     run_type=DagRunType.MANUAL,
                     logical_date=info.logical_date,
                     data_interval=info.data_interval,
+                    run_after=info.run_after,
                     triggered_by=DagRunTriggeredByType.TEST,
                     state=DagRunState.RUNNING,
                 )

airflow/models/dag.py

@@ -250,6 +250,7 @@ def _create_orm_dagrun(
     run_id: str,
     logical_date: datetime | None,
     data_interval: DataInterval | None,
+    run_after: datetime,
     start_date: datetime | None,
     external_trigger: bool,
     conf: Any,
@@ -266,6 +267,7 @@ def _create_orm_dagrun(
         run_id=run_id,
         logical_date=logical_date,
         start_date=start_date,
+        run_after=run_after,
         external_trigger=external_trigger,
         conf=conf,
         state=state,
@@ -1634,11 +1636,12 @@ def add_logger_if_needed(ti: TaskInstance):
                 dag=scheduler_dag,
                 start_date=logical_date,
                 logical_date=logical_date,
+                data_interval=data_interval,
+                run_after=data_interval.end,
                 run_id=DagRun.generate_run_id(DagRunType.MANUAL, logical_date),
                 session=session,
                 conf=run_conf,
                 triggered_by=DagRunTriggeredByType.TEST,
-                data_interval=data_interval,
             )
 
             tasks = self.task_dict
@@ -1721,6 +1724,7 @@ def create_dagrun(
         run_id: str,
         logical_date: datetime,
         data_interval: tuple[datetime, datetime],
+        run_after: datetime,
         conf: dict | None = None,
         run_type: DagRunType,
         triggered_by: DagRunTriggeredByType,
@@ -1791,6 +1795,8 @@ def create_dagrun(
             dag=self,
             run_id=run_id,
             logical_date=logical_date,
+            data_interval=data_interval,
+            run_after=timezone.coerce_datetime(run_after),
             start_date=timezone.coerce_datetime(start_date),
             external_trigger=external_trigger,
             conf=conf,
@@ -1799,7 +1805,6 @@ def create_dagrun(
             dag_version=dag_version,
             creating_job_id=creating_job_id,
             backfill_id=backfill_id,
-            data_interval=data_interval,
             triggered_by=triggered_by,
             session=session,
         )
@@ -2439,6 +2444,7 @@ def _get_or_create_dagrun(
     run_id: str,
     logical_date: datetime,
     data_interval: tuple[datetime, datetime],
+    run_after: datetime,
     conf: dict | None,
     triggered_by: DagRunTriggeredByType,
     start_date: datetime,
@@ -2468,6 +2474,7 @@ def _get_or_create_dagrun(
         run_id=run_id,
         logical_date=logical_date,
         data_interval=data_interval,
+        run_after=run_after,
         conf=conf,
         run_type=DagRunType.MANUAL,
         state=DagRunState.RUNNING,

airflow/models/dagrun.py

@@ -107,6 +107,11 @@ class TISchedulingDecision(NamedTuple):
     finished_tis: list[TI]
 
 
+def _default_run_after(ctx):
+    params = ctx.get_current_parameters()
+    return params["data_interval_end"] or params["logical_date"] or timezone.utcnow()
+
+
 def _creator_note(val):
     """Creator the ``note`` association proxy."""
     if isinstance(val, str):
@@ -145,6 +150,8 @@ class DagRun(Base, LoggingMixin):
     # These two must be either both NULL or both datetime.
     data_interval_start = Column(UtcDateTime)
     data_interval_end = Column(UtcDateTime)
+    # Earliest time when this DagRun can start running.
+    run_after = Column(UtcDateTime, default=_default_run_after, nullable=False)
     # When a scheduler last attempted to schedule TIs for this DagRun
     last_scheduling_decision = Column(UtcDateTime)
     # Foreign key to LogTemplate. DagRun rows created prior to this column's
@@ -180,6 +187,7 @@ class DagRun(Base, LoggingMixin):
         Index("dag_id_state", dag_id, _state),
         UniqueConstraint("dag_id", "run_id", name="dag_run_dag_id_run_id_key"),
         Index("idx_dag_run_dag_id", dag_id),
+        Index("idx_dag_run_run_after", run_after),
         Index(
             "idx_dag_run_running_dags",
             "state",
@@ -229,8 +237,10 @@ def __init__(
         self,
         dag_id: str | None = None,
         run_id: str | None = None,
+        *,
         queued_at: datetime | None | ArgNotSet = NOTSET,
         logical_date: datetime | None = None,
+        run_after: datetime | None = None,
         start_date: datetime | None = None,
         external_trigger: bool | None = None,
         conf: Any | None = None,
@@ -253,6 +263,7 @@ def __init__(
         self.dag_id = dag_id
         self.run_id = run_id
         self.logical_date = logical_date
+        self.run_after = run_after
         self.start_date = start_date
         self.external_trigger = external_trigger
         self.conf = conf or {}

airflow/utils/db.py

@@ -94,7 +94,7 @@ class MappedClassProtocol(Protocol):
     "2.9.2": "686269002441",
     "2.10.0": "22ed7efa9da2",
     "2.10.3": "5f2621c13b39",
-    "3.0.0": "33b04e4bfa19",
+    "3.0.0": "6a9e7a527a88",
 }
 
 

airflow/www/views.py

@@ -2237,6 +2237,7 @@ def trigger(self, dag_id: str, session: Session = NEW_SESSION):
                 run_id=run_id,
                 logical_date=logical_date,
                 data_interval=data_interval,
+                run_after=data_interval.end,
                 conf=run_conf,
                 run_type=DagRunType.MANUAL,
                 triggered_by=DagRunTriggeredByType.UI,

dev/perf/scheduler_dag_execution_timing.py

@@ -173,6 +173,7 @@ def create_dag_runs(dag, num_runs, session):
             run_id=f"{id_prefix}{logical_date.isoformat()}",
             logical_date=logical_date,
             data_interval=(logical_date, logical_date),
+            run_after=logical_date,
             run_type=DagRunType.MANUAL,
             triggered_by=DagRunTriggeredByType.TEST,
             external_trigger=False,

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-0f041321ecbb5d059f7a9c85fd5cfba745732eea205dc2e4f5e3b6dd76a9468d
+232f2f252ce0d3889fa5a9ceb00c88788e12083a6ea0c155c74d3fe61ad02412