AIP-83 Logical date should be required field when triggering run via API

airflow/api_fastapi/core_api/datamodels/dag_run.py

@@ -20,11 +20,11 @@
 from datetime import datetime
 from enum import Enum
 
-from pydantic import AwareDatetime, Field, NonNegativeInt, computed_field, model_validator
+import pendulum
+from pydantic import AwareDatetime, Field, NonNegativeInt, model_validator
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
 from airflow.models import DagRun
-from airflow.utils import timezone
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
@@ -82,6 +82,7 @@ class TriggerDAGRunPostBody(StrictBaseModel):
     """Trigger DAG Run Serializer for POST body."""
 
     dag_run_id: str | None = None
+    logical_date: AwareDatetime | None
     data_interval_start: AwareDatetime | None = None
     data_interval_end: AwareDatetime | None = None
 
@@ -99,15 +100,11 @@ def check_data_intervals(cls, values):
     @model_validator(mode="after")
     def validate_dag_run_id(self):
         if not self.dag_run_id:
-            self.dag_run_id = DagRun.generate_run_id(DagRunType.MANUAL, self.logical_date)
+            self.dag_run_id = DagRun.generate_run_id(
+                DagRunType.MANUAL, self.logical_date if self.logical_date is not None else pendulum.now("UTC")
+            )
         return self
 
-    # Mypy issue https://github.com/python/mypy/issues/1362
-    @computed_field  # type: ignore[misc]
-    @property
-    def logical_date(self) -> datetime:
-        return timezone.utcnow()
-
 
 class DAGRunsBatchBody(StrictBaseModel):
     """List DAG Runs body for batch endpoint."""

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -10291,6 +10291,12 @@ components:
           - type: string
           - type: 'null'
           title: Dag Run Id
+        logical_date:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Logical Date
         data_interval_start:
           anyOf:
           - type: string
@@ -10313,6 +10319,8 @@ components:
           title: Note
       additionalProperties: false
       type: object
+      required:
+      - logical_date
       title: TriggerDAGRunPostBody
       description: Trigger DAG Run Serializer for POST body.
     TriggerResponse:

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

@@ -351,7 +351,7 @@ def trigger_dag_run(
             f"DAG with dag_id: '{dag_id}' has import errors and cannot be triggered",
         )
 
-    logical_date = pendulum.instance(body.logical_date)
+    logical_date = pendulum.instance(body.logical_date) if body.logical_date is not None else None
 
     try:
         dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
@@ -362,16 +362,19 @@ def trigger_dag_run(
                 end=pendulum.instance(body.data_interval_end),
             )
         else:
-            data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date)
+            data_interval = dag.timetable.infer_manual_data_interval(
+                run_after=logical_date or pendulum.now("UTC")
+            )
 
-        if body.dag_run_id:
-            run_id = body.dag_run_id
-        else:
-            run_id = dag.timetable.generate_run_id(
+        run_id = (
+            body.dag_run_id
+            if body.dag_run_id
+            else dag.timetable.generate_run_id(
                 run_type=DagRunType.MANUAL,
-                logical_date=logical_date,
+                logical_date=logical_date or pendulum.now("UTC"),
                 data_interval=data_interval,
             )
+        )
 
         dag_run = dag.create_dagrun(
             run_id=run_id,

airflow/migrations/versions/0032_3_0_0_drop_execution_date_unique.py → airflow/migrations/versions/0032_3_0_0_rename_execution_date_to_logical_date_and_nullable.py

@@ -17,9 +17,9 @@
 # under the License.
 
 """
-Drop ``execution_date`` unique constraint on DagRun.
+Make logical_date nullable.
 
-The column has also been renamed to logical_date, although the Python model is
+The column has been renamed to logical_date, although the Python model is
 not changed. This allows us to not need to fix all the Python code at once, but
 still do the two changes in one migration instead of two.
 
@@ -49,10 +49,15 @@ def upgrade():
             "execution_date",
             new_column_name="logical_date",
             existing_type=TIMESTAMP(timezone=True),
-            existing_nullable=False,
+            existing_nullable=True,
         )
+
     with op.batch_alter_table("dag_run", schema=None) as batch_op:
         batch_op.drop_constraint("dag_run_dag_id_execution_date_key", type_="unique")
+        batch_op.create_unique_constraint(
+            "dag_run_dag_id_logical_date_key",
+            columns=["dag_id", "logical_date"],
+        )
 
 
 def downgrade():
@@ -63,7 +68,9 @@ def downgrade():
             existing_type=TIMESTAMP(timezone=True),
             existing_nullable=False,
         )
+
     with op.batch_alter_table("dag_run", schema=None) as batch_op:
+        batch_op.drop_constraint("dag_run_dag_id_logical_date_key", type_="unique")
         batch_op.create_unique_constraint(
             "dag_run_dag_id_execution_date_key",
             columns=["dag_id", "execution_date"],

airflow/models/backfill.py

@@ -1,4 +1,3 @@
-#
 # 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
@@ -141,6 +140,7 @@ class BackfillDagRunExceptionReason(str, Enum):
     IN_FLIGHT = "in flight"
     ALREADY_EXISTS = "already exists"
     UNKNOWN = "unknown"
+    CLEARED_RUN = "cleared existing run"
 
 
 class BackfillDagRun(Base):
@@ -194,7 +194,11 @@ def _get_latest_dag_run_row_query(info, session):
 def _get_dag_run_no_create_reason(dr, reprocess_behavior: ReprocessBehavior) -> str | None:
     non_create_reason = None
     if dr.state not in (DagRunState.SUCCESS, DagRunState.FAILED):
-        non_create_reason = BackfillDagRunExceptionReason.IN_FLIGHT
+        if dr.clear_number == 0:
+            non_create_reason = BackfillDagRunExceptionReason.IN_FLIGHT
+        else:
+            non_create_reason = BackfillDagRunExceptionReason.CLEARED_RUN
+
     elif reprocess_behavior is ReprocessBehavior.NONE:
         non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
     elif reprocess_behavior is ReprocessBehavior.FAILED:
@@ -262,7 +266,23 @@ def _create_backfill_dag_run(
     dag_run_conf,
     backfill_sort_ordinal,
     session,
+    from_date,
+    to_date,
 ):
+    from airflow.models import DAG
+
+    dr = session.scalar(_get_latest_dag_run_row_query(info, session))
+    if (
+        dr
+        and dr.state not in {DagRunState.RUNNING}
+        and reprocess_behavior in {ReprocessBehavior.COMPLETED, ReprocessBehavior.FAILED}
+    ):
+        DAG.clear_dags(
+            [dag],
+            start_date=from_date,
+            end_date=to_date,
+            dag_run_state=DagRunState.QUEUED,
+        )
     with session.begin_nested() as nested:
         dr = session.scalar(
             with_row_locks(
@@ -412,6 +432,8 @@ def _create_backfill(
                 reprocess_behavior=br.reprocess_behavior,
                 backfill_sort_ordinal=backfill_sort_ordinal,
                 session=session,
+                from_date=from_date,
+                to_date=to_date,
             )
             log.info(
                 "created backfill dag run dag_id=%s backfill_id=%s, info=%s",

airflow/models/dag.py

@@ -1722,7 +1722,7 @@ def create_dagrun(
         self,
         *,
         run_id: str,
-        logical_date: datetime,
+        logical_date: datetime | None,
         data_interval: tuple[datetime, datetime],
         run_after: datetime,
         conf: dict | None = None,
@@ -1747,7 +1747,8 @@ def create_dagrun(
 
         :meta private:
         """
-        logical_date = timezone.coerce_datetime(logical_date)
+        if logical_date is not None:
+            logical_date = timezone.coerce_datetime(logical_date)
 
         if data_interval and not isinstance(data_interval, DataInterval):
             data_interval = DataInterval(*map(timezone.coerce_datetime, data_interval))

airflow/models/dagrun.py

@@ -135,7 +135,7 @@ class DagRun(Base, LoggingMixin):
     id = Column(Integer, primary_key=True)
     dag_id = Column(StringID(), nullable=False)
     queued_at = Column(UtcDateTime)
-    logical_date = Column(UtcDateTime, default=timezone.utcnow, nullable=False)
+    logical_date = Column(UtcDateTime, nullable=True)
     start_date = Column(UtcDateTime)
     end_date = Column(UtcDateTime)
     _state = Column("state", String(50), default=DagRunState.QUEUED)
@@ -186,6 +186,7 @@ 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"),
         Index("idx_dag_run_dag_id", dag_id),
         Index("idx_dag_run_run_after", run_after),
         Index(
@@ -1315,8 +1316,12 @@ def verify_integrity(self, *, session: Session = NEW_SESSION) -> None:
         def task_filter(task: Operator) -> bool:
             return task.task_id not in task_ids and (
                 self.run_type == DagRunType.BACKFILL_JOB
-                or (task.start_date is None or task.start_date <= self.logical_date)
-                and (task.end_date is None or self.logical_date <= task.end_date)
+                or (
+                    task.start_date is None
+                    or self.logical_date is None
+                    or task.start_date <= self.logical_date
+                )
+                and (task.end_date is None or self.logical_date is None or self.logical_date <= task.end_date)
             )
 
         created_counts: dict[str, int] = defaultdict(int)

airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -5752,6 +5752,18 @@ export const $TriggerDAGRunPostBody = {
       ],
       title: "Dag Run Id",
     },
+    logical_date: {
+      anyOf: [
+        {
+          type: "string",
+          format: "date-time",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Logical Date",
+    },
     data_interval_start: {
       anyOf: [
         {
@@ -5794,6 +5806,7 @@ export const $TriggerDAGRunPostBody = {
   },
   additionalProperties: false,
   type: "object",
+  required: ["logical_date"],
   title: "TriggerDAGRunPostBody",
   description: "Trigger DAG Run Serializer for POST body.",
 } as const;

airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1402,6 +1402,7 @@ export type TimeDelta = {
  */
 export type TriggerDAGRunPostBody = {
   dag_run_id?: string | null;
+  logical_date: string | null;
   data_interval_start?: string | null;
   data_interval_end?: string | null;
   conf?: {

airflow/ui/src/queries/useTrigger.ts

@@ -109,6 +109,7 @@ export const useTrigger = ({ dagId, onSuccessConfirm }: { dagId: string; onSucce
         dag_run_id: checkDagRunId,
         data_interval_end: formattedDataIntervalEnd,
         data_interval_start: formattedDataIntervalStart,
+        logical_date: null,
         note: checkNote,
       },
     });

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-232f2f252ce0d3889fa5a9ceb00c88788e12083a6ea0c155c74d3fe61ad02412
+c27770b242bc31b945e731d3b82844c02a1a9a4974061e55af1d957985741881

docs/apache-airflow/migrations-ref.rst

@@ -92,7 +92,7 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``522625f6d606``        | ``1cdc775ca98f`` | ``3.0.0``         | Add tables for backfill.                                     |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``1cdc775ca98f``        | ``a2c32e6c7729`` | ``3.0.0``         | Drop ``execution_date`` unique constraint on DagRun.         |
+| ``1cdc775ca98f``        | ``a2c32e6c7729`` | ``3.0.0``         | Make logical_date nullable.                                  |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``a2c32e6c7729``        | ``0bfc26bc256e`` | ``3.0.0``         | Add triggered_by field to DagRun.                            |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

tests/api_fastapi/core_api/routes/public/test_assets.py

@@ -149,7 +149,7 @@ def _create_dag_run(session, num: int = 2):
             dag_id="source_dag_id",
             run_id=f"source_run_id_{i}",
             run_type=DagRunType.MANUAL,
-            logical_date=DEFAULT_DATE,
+            logical_date=DEFAULT_DATE + timedelta(days=i),
             start_date=DEFAULT_DATE,
             data_interval=(DEFAULT_DATE, DEFAULT_DATE),
             external_trigger=True,
@@ -554,7 +554,9 @@ def test_should_respond_200(self, test_client, session):
                         {
                             "run_id": "source_run_id_1",
                             "dag_id": "source_dag_id",
-                            "logical_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
+                            "logical_date": from_datetime_to_zulu_without_ms(
+                                DEFAULT_DATE + timedelta(days=1)
+                            ),
                             "start_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "end_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "state": "success",
@@ -579,7 +581,9 @@ def test_should_respond_200(self, test_client, session):
                         {
                             "run_id": "source_run_id_2",
                             "dag_id": "source_dag_id",
-                            "logical_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
+                            "logical_date": from_datetime_to_zulu_without_ms(
+                                DEFAULT_DATE + timedelta(days=2)
+                            ),
                             "start_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "end_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "state": "success",
@@ -722,7 +726,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
                         {
                             "run_id": "source_run_id_1",
                             "dag_id": "source_dag_id",
-                            "logical_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
+                            "logical_date": from_datetime_to_zulu_without_ms(
+                                DEFAULT_DATE + timedelta(days=1)
+                            ),
                             "start_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "end_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "state": "success",
@@ -747,7 +753,9 @@ def test_should_mask_sensitive_extra(self, test_client, session):
                         {
                             "run_id": "source_run_id_2",
                             "dag_id": "source_dag_id",
-                            "logical_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
+                            "logical_date": from_datetime_to_zulu_without_ms(
+                                DEFAULT_DATE + timedelta(days=2)
+                            ),
                             "start_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "end_date": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
                             "state": "success",