Repository: apache/airflow
Pull request: Handle future dates cases in Backfill
Closes: https://github.com/apache/airflow/issues/46012
This PR fixes the below behaviors
1. Ensure dates which are in future with run_backwards=true are not inserted in backfilldagrun table
2. Raise 422 when all dates are in the future.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).


Problem statement:
backfill should not create runs for future dates
### Body


 backfill shouldn't be created for future dates I believe but if I pass run_backwards=true then its creating backfill for future date also
reverse block is missing the condition if x.data_interval.end < now
Also we can reuse the infos in the reverse block
def _get_info_list(
    *,
    from_date,
    to_date,
    reverse,
    dag,
):
    infos = dag.iter_dagrun_infos_between(from_date, to_date)
    now = timezone.utcnow()
    dagrun_info_list = (x for x in infos if x.data_interval.end < now)
    if reverse:
        dagrun_info_list = reversed([x for x in dag.iter_dagrun_infos_between(from_date, to_date)])
    return dagrun_info_list

### Committer

- [x] I acknowledge that I am a maintainer/committer of the Apache Airflow project.

Hints from issue discussion:

Commit message: fix backills future date behaviour

Patch to review:
diff --git a/airflow/api_fastapi/core_api/routes/public/backfills.py b/airflow/api_fastapi/core_api/routes/public/backfills.py
index 78f0391fec975..7c4dc9d32a2d0 100644
--- a/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -46,6 +46,7 @@
     Backfill,
     BackfillDagRun,
     DagNoScheduleException,
+    InvalidBackfillDate,
     InvalidBackfillDirection,
     InvalidReprocessBehavior,
     _create_backfill,
@@ -220,6 +221,7 @@ def create_backfill(
         InvalidReprocessBehavior,
         InvalidBackfillDirection,
         DagNoScheduleException,
+        InvalidBackfillDate,
     ) as e:
         raise RequestValidationError(str(e))
 
@@ -254,5 +256,10 @@ def create_backfill_dry_run(
             detail=f"Could not find dag {body.dag_id}",
         )
 
-    except (InvalidReprocessBehavior, InvalidBackfillDirection, DagNoScheduleException) as e:
+    except (
+        InvalidReprocessBehavior,
+        InvalidBackfillDirection,
+        DagNoScheduleException,
+        InvalidBackfillDate,
+    ) as e:
         raise RequestValidationError(str(e))
diff --git a/airflow/models/backfill.py b/airflow/models/backfill.py
index 1564c4feb59a5..98ce9ce20fae6 100644
--- a/airflow/models/backfill.py
+++ b/airflow/models/backfill.py
@@ -91,6 +91,14 @@ class InvalidReprocessBehavior(AirflowException):
     """
 
 
+class InvalidBackfillDate(AirflowException):
+    """
+    Raised when a backfill is requested for future date.
+
+    :meta private:
+    """
+
+
 class ReprocessBehavior(str, Enum):
     """
     Internal enum for setting reprocess behavior in a backfill.
@@ -203,7 +211,7 @@ def _get_dag_run_no_create_reason(dr, reprocess_behavior: ReprocessBehavior) ->
     return non_create_reason
 
 
-def _validate_backfill_params(dag, reverse, reprocess_behavior: ReprocessBehavior | None):
+def _validate_backfill_params(dag, reverse, from_date, to_date, reprocess_behavior: ReprocessBehavior | None):
     depends_on_past = any(x.depends_on_past for x in dag.tasks)
     if depends_on_past:
         if reverse is True:
@@ -215,6 +223,8 @@ def _validate_backfill_params(dag, reverse, reprocess_behavior: ReprocessBehavio
                 "DAG has tasks for which depends_on_past=True. "
                 "You must set reprocess behavior to reprocess completed or reprocess failed."
             )
+    if from_date >= timezone.utcnow() and to_date >= timezone.utcnow():
+        raise InvalidBackfillDate("Backfill cannot be executed for future dates.")
 
 
 def _do_dry_run(*, dag_id, from_date, to_date, reverse, reprocess_behavior, session) -> list[datetime]:
@@ -225,7 +235,7 @@ def _do_dry_run(*, dag_id, from_date, to_date, reverse, reprocess_behavior, sess
     if not serdag:
         raise DagNotFound(f"Could not find dag {dag_id}")
     dag = serdag.dag
-    _validate_backfill_params(dag, reverse, reprocess_behavior)
+    _validate_backfill_params(dag, reverse, from_date, to_date, reprocess_behavior)
 
     no_schedule = session.scalar(
         select(func.count()).where(DagModel.timetable_summary == "None", DagModel.dag_id == dag_id)
@@ -326,7 +336,7 @@ def _get_info_list(
     now = timezone.utcnow()
     dagrun_info_list = (x for x in infos if x.data_interval.end < now)
     if reverse:
-        dagrun_info_list = reversed([x for x in dag.iter_dagrun_infos_between(from_date, to_date)])
+        dagrun_info_list = reversed([x for x in infos if x.data_interval.end < now])
     return dagrun_info_list
 
 
@@ -366,7 +376,7 @@ def _create_backfill(
             )
 
         dag = serdag.dag
-        _validate_backfill_params(dag, reverse, reprocess_behavior)
+        _validate_backfill_params(dag, reverse, from_date, to_date, reprocess_behavior)
 
         br = Backfill(
             dag_id=dag_id,
diff --git a/tests/api_fastapi/core_api/routes/public/test_backfills.py b/tests/api_fastapi/core_api/routes/public/test_backfills.py
index 0a71963c1007f..b6bd12f38732b 100644
--- a/tests/api_fastapi/core_api/routes/public/test_backfills.py
+++ b/tests/api_fastapi/core_api/routes/public/test_backfills.py
@@ -17,12 +17,12 @@
 from __future__ import annotations
 
 import os
-from datetime import datetime
+from datetime import datetime, timedelta
 from unittest import mock
 
 import pendulum
 import pytest
-from sqlalchemy import select
+from sqlalchemy import and_, func, select
 
 from airflow.models import DagBag, DagModel, DagRun
 from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior, _create_backfill
@@ -315,6 +315,72 @@ def test_create_backfill_with_depends_on_past(
                     == "DAG has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
                 )
 
+    @pytest.mark.parametrize(
+        "run_backwards",
+        [
+            (False),
+            (True),
+        ],
+    )
+    def test_create_backfill_future_dates(self, session, dag_maker, test_client, run_backwards):
+        with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+            EmptyOperator(task_id="mytask")
+        session.query(DagModel).all()
+        session.commit()
+        from_date = timezone.utcnow() + timedelta(days=1)
+        to_date = timezone.utcnow() + timedelta(days=1)
+        max_active_runs = 5
+        data = {
+            "dag_id": dag.dag_id,
+            "from_date": f"{to_iso(from_date)}",
+            "to_date": f"{to_iso(to_date)}",
+            "max_active_runs": max_active_runs,
+            "run_backwards": run_backwards,
+            "dag_run_conf": {"param1": "val1", "param2": True},
+        }
+
+        response = test_client.post(
+            url="/public/backfills",
+            json=data,
+        )
+        assert response.status_code == 422
+        assert response.json().get("detail") == "Backfill cannot be executed for future dates."
+
+    @pytest.mark.parametrize(
+        "run_backwards",
+        [
+            (False),
+            (True),
+        ],
+    )
+    def test_create_backfill_past_future_dates(self, session, dag_maker, test_client, run_backwards):
+        with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="@daily") as dag:
+            EmptyOperator(task_id="mytask")
+        session.query(DagModel).all()
+        session.commit()
+        from_date = timezone.utcnow() - timedelta(days=2)
+        to_date = timezone.utcnow() + timedelta(days=1)
+        max_active_runs = 1
+        data = {
+            "dag_id": dag.dag_id,
+            "from_date": f"{to_iso(from_date)}",
+            "to_date": f"{to_iso(to_date)}",
+            "max_active_runs": max_active_runs,
+            "run_backwards": run_backwards,
+            "dag_run_conf": {"param1": "val1", "param2": True},
+        }
+
+        response = test_client.post(
+            url="/public/backfills",
+            json=data,
+        )
+        assert response.status_code == 200
+        backfill_dag_run_count = select(func.count()).where(
+            and_(BackfillDagRun.backfill_id == response.json()["id"], BackfillDagRun.logical_date == to_date)
+        )
+        count = session.execute(backfill_dag_run_count).scalar()
+        assert count == 0
+
 
 class TestCreateBackfillDryRun(TestBackfillEndpoint):
     @pytest.mark.parametrize(


Code review comments:
diff_hunk: @@ -215,6 +223,8 @@ def _validate_backfill_params(dag, reverse, reprocess_behavior: ReprocessBehavio
                 "DAG has tasks for which depends_on_past=True. "
                 "You must set reprocess behavior to reprocess completed or reprocess failed."
             )
+    if from_date >= timezone.utcnow() and to_date >= timezone.utcnow():
path: airflow/models/backfill.py
text: @user1:
timezone.utcnow() can be extracted to the same constant value. instead of called twice.

@user1:
What if only one of the two dates (from_date or to_date) is in the future ? Should that be considered invalid or valid ? 

@author:
I think to_date is already being handled [here](https://github.com/apache/airflow/pull/46017/files#diff-f964a968e76ed62e5a64804b6c2387c790ec8ddc8aeeaced41cdc4b43413db06R338)