Add dry run for backfill

airflow/api_fastapi/core_api/datamodels/backfills.py

@@ -56,3 +56,16 @@ class BackfillCollectionResponse(BaseModel):
 
     backfills: list[BackfillResponse]
     total_entries: int
+
+
+class DryRunBackfillResponse(BaseModel):
+    """Backfill serializer for responses in dry-run mode."""
+
+    logical_date: datetime
+
+
+class DryRunBackfillCollectionResponse(BaseModel):
+    """Backfill collection serializer for responses in dry-run mode."""
+
+    backfills: list[DryRunBackfillResponse]
+    total_entries: int

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

@@ -1481,6 +1481,55 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /public/backfills/dry_run:
+    post:
+      tags:
+      - Backfill
+      summary: Create Backfill Dry Run
+      operationId: create_backfill_dry_run
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/BackfillPostBody'
+        required: true
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/DryRunBackfillCollectionResponse'
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '403':
+          description: Forbidden
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '404':
+          description: Not Found
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '409':
+          description: Conflict
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /public/connections/{connection_id}:
     delete:
       tags:
@@ -7964,6 +8013,33 @@ components:
         This is the set of allowable values for the ``warning_type`` field
 
         in the DagWarning model.'
+    DryRunBackfillCollectionResponse:
+      properties:
+        backfills:
+          items:
+            $ref: '#/components/schemas/DryRunBackfillResponse'
+          type: array
+          title: Backfills
+        total_entries:
+          type: integer
+          title: Total Entries
+      type: object
+      required:
+      - backfills
+      - total_entries
+      title: DryRunBackfillCollectionResponse
+      description: Backfill collection serializer for responses in dry-run mode.
+    DryRunBackfillResponse:
+      properties:
+        logical_date:
+          type: string
+          format: date-time
+          title: Logical Date
+      type: object
+      required:
+      - logical_date
+      title: DryRunBackfillResponse
+      description: Backfill serializer for responses in dry-run mode.
     EdgeResponse:
       properties:
         is_setup_teardown:

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

@@ -32,6 +32,8 @@
     BackfillCollectionResponse,
     BackfillPostBody,
     BackfillResponse,
+    DryRunBackfillCollectionResponse,
+    DryRunBackfillResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
@@ -42,6 +44,7 @@
     Backfill,
     BackfillDagRun,
     _create_backfill,
+    _do_dry_run,
 )
 from airflow.utils import timezone
 from airflow.utils.state import DagRunState
@@ -206,3 +209,30 @@ def create_backfill(
             status_code=status.HTTP_409_CONFLICT,
             detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
         )
+
+
+@backfills_router.post(
+    path="/dry_run",
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_404_NOT_FOUND,
+            status.HTTP_409_CONFLICT,
+        ]
+    ),
+)
+def create_backfill_dry_run(
+    body: BackfillPostBody,
+) -> DryRunBackfillCollectionResponse:
+    from_date = timezone.coerce_datetime(body.from_date)
+    to_date = timezone.coerce_datetime(body.to_date)
+
+    backfills_dry_run = _do_dry_run(
+        dag_id=body.dag_id,
+        from_date=from_date,
+        to_date=to_date,
+        reverse=body.run_backwards,
+        reprocess_behavior=body.reprocess_behavior,
+    )
+    backfills = [DryRunBackfillResponse(logical_date=d) for d in backfills_dry_run]
+
+    return DryRunBackfillCollectionResponse(backfills=backfills, total_entries=len(backfills_dry_run))

airflow/cli/commands/remote_commands/backfill_command.py

@@ -21,31 +21,10 @@
 import signal
 
 from airflow import settings
-from airflow.models.backfill import ReprocessBehavior, _create_backfill, _get_info_list
-from airflow.models.serialized_dag import SerializedDagModel
+from airflow.models.backfill import ReprocessBehavior, _create_backfill, _do_dry_run
 from airflow.utils import cli as cli_utils
 from airflow.utils.cli import sigint_handler
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
-from airflow.utils.session import create_session
-
-
-def _do_dry_run(*, params, dag_id, from_date, to_date, reverse):
-    print("Performing dry run of backfill.")
-    print("Printing params:")
-    for k, v in params.items():
-        print(f"    - {k} = {v}")
-    with create_session() as session:
-        serdag = session.get(SerializedDagModel, dag_id)
-
-    info_list = _get_info_list(
-        dag=serdag.dag,
-        from_date=from_date,
-        to_date=to_date,
-        reverse=reverse,
-    )
-    print("Logical dates to be attempted:")
-    for info in info_list:
-        print(f"    - {info.logical_date}")
 
 
 @cli_utils.action_cli
@@ -61,22 +40,31 @@ def create_backfill(args) -> None:
         reprocess_behavior = None
 
     if args.dry_run:
-        _do_dry_run(
-            params=dict(
-                dag_id=args.dag_id,
-                from_date=args.from_date,
-                to_date=args.to_date,
-                max_active_runs=args.max_active_runs,
-                reverse=args.run_backwards,
-                dag_run_conf=args.dag_run_conf,
-                reprocess_behavior=reprocess_behavior,
-            ),
+        print("Performing dry run of backfill.")
+        print("Printing params:")
+        params = dict(
             dag_id=args.dag_id,
             from_date=args.from_date,
             to_date=args.to_date,
+            max_active_runs=args.max_active_runs,
             reverse=args.run_backwards,
+            dag_run_conf=args.dag_run_conf,
+            reprocess_behavior=reprocess_behavior,
         )
+        for k, v in params.items():
+            print(f"    - {k} = {v}")
+        logical_dates = _do_dry_run(
+            dag_id=args.dag_id,
+            from_date=args.from_date,
+            to_date=args.to_date,
+            reverse=args.reverse,
+            reprocess_behavior=args.reprocess_behavior,
+        )
+        print("Logical dates to be attempted:")
+        for d in logical_dates:
+            print(f"    - {d}")
         return
+
     _create_backfill(
         dag_id=args.dag_id,
         from_date=args.from_date,

airflow/models/backfill.py

@@ -155,6 +155,73 @@ def validate_sort_ordinal(self, key, val):
         return val
 
 
+def _get_latest_dag_run_row_query(info, session):
+    from airflow.models import DagRun
+
+    return (
+        select(DagRun)
+        .where(DagRun.logical_date == info.logical_date)
+        .order_by(nulls_first(desc(DagRun.start_date), session=session))
+        .limit(1)
+    )
+
+
+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
+    elif reprocess_behavior is ReprocessBehavior.NONE:
+        non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
+    elif reprocess_behavior is ReprocessBehavior.FAILED:
+        if dr.state != DagRunState.FAILED:
+            non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
+    return non_create_reason
+
+
+def _validate_backfill_params(dag, reverse, reprocess_behavior: ReprocessBehavior | None):
+    depends_on_past = None
+    depends_on_past = any(x.depends_on_past for x in dag.tasks)
+    if depends_on_past:
+        if reverse is True:
+            raise ValueError(
+                "Backfill cannot be run in reverse when the dag has tasks where depends_on_past=True"
+            )
+        if reprocess_behavior in (None, ReprocessBehavior.NONE):
+            raise ValueError(
+                "Dag has task for which depends_on_past is true. "
+                "You must set reprocess behavior to reprocess completed or "
+                "reprocess failed"
+            )
+
+
+def _do_dry_run(*, dag_id, from_date, to_date, reverse, reprocess_behavior) -> list[datetime]:
+    from airflow.models.serialized_dag import SerializedDagModel
+
+    with create_session() as session:
+        serdag = session.scalar(SerializedDagModel.latest_item_select_object(dag_id))
+        dag = serdag.dag
+        _validate_backfill_params(dag, reverse, reprocess_behavior)
+
+        dagrun_info_list = _get_info_list(
+            dag=dag,
+            from_date=from_date,
+            to_date=to_date,
+            reverse=reverse,
+        )
+        logical_dates = []
+        for info in dagrun_info_list:
+            dr = session.scalar(
+                statement=_get_latest_dag_run_row_query(info, session),
+            )
+            if dr:
+                non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
+                if not non_create_reason:
+                    logical_dates.append(info.logical_date)
+            else:
+                logical_dates.append(info.logical_date)
+        return logical_dates
+
+
 def _create_backfill_dag_run(
     *,
     dag,
@@ -165,27 +232,15 @@ def _create_backfill_dag_run(
     backfill_sort_ordinal,
     session,
 ):
-    from airflow.models import DagRun
-
     with session.begin_nested() as nested:
         dr = session.scalar(
             with_row_locks(
-                select(DagRun)
-                .where(DagRun.logical_date == info.logical_date)
-                .order_by(nulls_first(desc(DagRun.start_date), session=session))
-                .limit(1),
+                query=_get_latest_dag_run_row_query(info, session),
                 session=session,
-            )
+            ),
         )
         if dr:
-            non_create_reason = None
-            if dr.state not in (DagRunState.SUCCESS, DagRunState.FAILED):
-                non_create_reason = BackfillDagRunExceptionReason.IN_FLIGHT
-            elif reprocess_behavior is ReprocessBehavior.NONE:
-                non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
-            elif reprocess_behavior is ReprocessBehavior.FAILED:
-                if dr.state != DagRunState.FAILED:
-                    non_create_reason = BackfillDagRunExceptionReason.ALREADY_EXISTS
+            non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
             if non_create_reason:
                 # rolling back here restores to start of this nested tran
                 # which releases the lock on the latest dag run, since we
@@ -272,18 +327,8 @@ def _create_backfill(
             )
 
         dag = serdag.dag
-        depends_on_past = any(x.depends_on_past for x in dag.tasks)
-        if depends_on_past:
-            if reverse is True:
-                raise ValueError(
-                    "Backfill cannot be run in reverse when the dag has tasks where depends_on_past=True"
-                )
-            if reprocess_behavior in (None, ReprocessBehavior.NONE):
-                raise ValueError(
-                    "Dag has task for which depends_on_past is true. "
-                    "You must set reprocess behavior to reprocess completed or "
-                    "reprocess failed"
-                )
+        _validate_backfill_params(dag, reverse, reprocess_behavior)
+
         br = Backfill(
             dag_id=dag_id,
             from_date=from_date,
@@ -316,6 +361,7 @@ def _create_backfill(
         )
         if not dag_model:
             raise RuntimeError(f"Dag {dag_id} not found")
+
         for info in dagrun_info_list:
             backfill_sort_ordinal += 1
             _create_backfill_dag_run(

airflow/ui/openapi-gen/queries/common.ts

@@ -1596,6 +1596,9 @@ export type AssetServiceCreateAssetEventMutationResult = Awaited<
 export type BackfillServiceCreateBackfillMutationResult = Awaited<
   ReturnType<typeof BackfillService.createBackfill>
 >;
+export type BackfillServiceCreateBackfillDryRunMutationResult = Awaited<
+  ReturnType<typeof BackfillService.createBackfillDryRun>
+>;
 export type ConnectionServicePostConnectionMutationResult = Awaited<
   ReturnType<typeof ConnectionService.postConnection>
 >;