You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by we...@apache.org on 2026/06/30 09:01:06 UTC
(airflow) branch backfill-empty-window-ux created (now 3e4fb93f012)
This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a change to branch backfill-empty-window-ux
in repository https://gitbox.apache.org/repos/asf/airflow.git
at 3e4fb93f012 Return 422 for empty partitioned backfill window and fix dry-run drift
This branch includes the following new commits:
new 3e4fb93f012 Return 422 for empty partitioned backfill window and fix dry-run drift
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: Return 422 for empty partitioned backfill window and fix dry-run drift
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch backfill-empty-window-ux
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3e4fb93f0129e18a4d5978134b91a177d2df59d3
Author: Wei Lee <we...@gmail.com>
AuthorDate: Mon Jun 22 23:01:25 2026 +0800
Return 422 for empty partitioned backfill window and fix dry-run drift
A partitioned backfill whose window contains no partition runs (e.g. a
sparse weekly timetable asked to backfill a single non-matching day) used
to surface as a raw HTTP 500, and the create path committed the Backfill
row before checking for runs — leaving an active orphan that blocked every
later retry with AlreadyRunningBackfill.
Empty windows are now detected before any DB write and raised as
NoBackfillRunsToCreate: the REST route returns 422 and the CLI prints a
friendly message instead of a traceback. Dry-run keeps returning zero
entries (a valid preview result), so create and preview stay intentionally
asymmetric.
The CLI dry-run path now converts reprocess_behavior to its enum and passes
dag_run_conf, matching what the real create path receives so the preview no
longer drifts from execution.
---
.../core_api/routes/public/backfills.py | 2 +
.../src/airflow/cli/commands/backfill_command.py | 48 ++++++------
airflow-core/src/airflow/models/backfill.py | 31 +++++---
.../core_api/routes/public/test_backfills.py | 58 +++++++++++++-
.../unit/cli/commands/test_backfill_command.py | 61 ++++++++++++++-
airflow-core/tests/unit/models/test_backfill.py | 90 ++++++++++++++++++++++
6 files changed, 254 insertions(+), 36 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
index 0dba6086e0a..70c353fb7a2 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -56,6 +56,7 @@ from airflow.models.backfill import (
InvalidBackfillDateRange,
InvalidBackfillDirection,
InvalidReprocessBehavior,
+ NoBackfillRunsToCreate,
_create_backfill,
_do_dry_run,
)
@@ -276,6 +277,7 @@ def create_backfill(
InvalidBackfillDate,
InvalidBackfillDateRange,
InvalidBackfillConf,
+ NoBackfillRunsToCreate,
) as e:
raise RequestValidationError(str(e))
diff --git a/airflow-core/src/airflow/cli/commands/backfill_command.py b/airflow-core/src/airflow/cli/commands/backfill_command.py
index d76d435a391..75ba6b03ce4 100644
--- a/airflow-core/src/airflow/cli/commands/backfill_command.py
+++ b/airflow-core/src/airflow/cli/commands/backfill_command.py
@@ -27,7 +27,7 @@ from airflow import settings
from airflow.api_fastapi.common.dagbag import resolve_run_on_latest_version
from airflow.cli.simple_table import AirflowConsole
from airflow.exceptions import AirflowConfigException
-from airflow.models.backfill import ReprocessBehavior, _create_backfill, _do_dry_run
+from airflow.models.backfill import NoBackfillRunsToCreate, ReprocessBehavior, _create_backfill, _do_dry_run
from airflow.utils import cli as cli_utils
from airflow.utils.cli import sigint_handler
from airflow.utils.platform import getuser
@@ -50,6 +50,13 @@ def create_backfill(args) -> None:
else:
reprocess_behavior = None
+ dag_run_conf = None
+ if args.dag_run_conf:
+ try:
+ dag_run_conf = json.loads(args.dag_run_conf)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Invalid JSON in --dag-run-conf: {e}")
+
with create_session() as session:
resolved_run_on_latest = resolve_run_on_latest_version(
args.run_on_latest_version,
@@ -67,7 +74,7 @@ def create_backfill(args) -> None:
to_date=args.to_date,
max_active_runs=args.max_active_runs,
reverse=args.run_backwards,
- dag_run_conf=args.dag_run_conf,
+ dag_run_conf=dag_run_conf,
reprocess_behavior=reprocess_behavior,
run_on_latest_version=resolved_run_on_latest,
)
@@ -79,7 +86,8 @@ def create_backfill(args) -> None:
from_date=args.from_date,
to_date=args.to_date,
reverse=args.run_backwards,
- reprocess_behavior=args.reprocess_behavior,
+ reprocess_behavior=reprocess_behavior or ReprocessBehavior.NONE,
+ dag_run_conf=dag_run_conf,
session=session,
)
console.print("Runs to be attempted:")
@@ -97,22 +105,18 @@ def create_backfill(args) -> None:
log.warning("Failed to get user name from os: %s, not setting the triggering user", e)
user = None
- # Parse dag_run_conf if provided
- dag_run_conf = None
- if args.dag_run_conf:
- try:
- dag_run_conf = json.loads(args.dag_run_conf)
- except json.JSONDecodeError as e:
- raise ValueError(f"Invalid JSON in --dag-run-conf: {e}")
-
- _create_backfill(
- 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=dag_run_conf,
- triggering_user_name=user,
- reprocess_behavior=reprocess_behavior,
- run_on_latest_version=resolved_run_on_latest,
- )
+ try:
+ _create_backfill(
+ 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=dag_run_conf,
+ triggering_user_name=user,
+ reprocess_behavior=reprocess_behavior,
+ run_on_latest_version=resolved_run_on_latest,
+ )
+ except NoBackfillRunsToCreate as e:
+ console.print(f"[yellow]Warning:[/yellow] {e}")
+ raise SystemExit(1)
diff --git a/airflow-core/src/airflow/models/backfill.py b/airflow-core/src/airflow/models/backfill.py
index 52019fa5d98..1f507de94bb 100644
--- a/airflow-core/src/airflow/models/backfill.py
+++ b/airflow-core/src/airflow/models/backfill.py
@@ -123,6 +123,17 @@ class InvalidBackfillConf(AirflowException):
"""
+class NoBackfillRunsToCreate(ValueError):
+ """
+ Raised when a backfill request yields no Dag runs for the given date range.
+
+ This happens when the from/to date range falls entirely outside the Dag's
+ scheduled intervals (e.g. the range predates the first partition boundary).
+
+ :meta private:
+ """
+
+
class UnknownActiveBackfills(AirflowException):
"""
Raised when the quantity of active backfills cannot be determined.
@@ -665,6 +676,17 @@ def _create_backfill(
dag_run_conf,
)
+ dagrun_info_list = _get_info_list(
+ from_date=from_date,
+ to_date=to_date,
+ reverse=reverse,
+ dag=dag,
+ )
+ if not dagrun_info_list:
+ raise NoBackfillRunsToCreate(
+ f"No runs to create for Dag {dag_id} in the range [{from_date}, {to_date}]"
+ )
+
br = Backfill(
dag_id=dag_id,
from_date=from_date,
@@ -680,15 +702,6 @@ def _create_backfill(
session.scalars(select(DagModel).where(DagModel.dag_id == dag_id)).one()
- dagrun_info_list = _get_info_list(
- from_date=from_date,
- to_date=to_date,
- reverse=reverse,
- dag=dag,
- )
- if not dagrun_info_list:
- raise RuntimeError(f"No runs to create for Dag {dag_id}")
-
first_info = dagrun_info_list[0]
if first_info.partition_key:
_create_runs_partitioned(
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
index 42b48fc0c49..2380f73b9bf 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -27,7 +27,13 @@ from sqlalchemy import and_, func, select
from airflow._shared.timezones import timezone
from airflow.dag_processing.dagbag import DagBag
from airflow.models import DagModel, DagRun
-from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior, _create_backfill
+from airflow.models.backfill import (
+ Backfill,
+ BackfillDagRun,
+ NoBackfillRunsToCreate,
+ ReprocessBehavior,
+ _create_backfill,
+)
from airflow.models.dag import DAG
from airflow.models.dagbundle import DagBundleModel
from airflow.providers.standard.operators.empty import EmptyOperator
@@ -965,6 +971,56 @@ class TestCreateBackfillPartitioned(TestBackfillEndpoint):
response = test_client.post(url=url, json=data)
assert response.status_code == 422
+ @mock.patch("airflow.api_fastapi.core_api.routes.public.backfills._create_backfill", autospec=True)
+ def test_empty_window_create_returns_422(self, mock_create, session, dag_maker, test_client):
+ """POST /backfills with an empty window raises NoBackfillRunsToCreate → 422."""
+ mock_create.side_effect = NoBackfillRunsToCreate(
+ "No runs to create for Dag TEST_PARTITIONED_DAG in the range [...]"
+ )
+ with dag_maker(
+ session=session,
+ dag_id="TEST_PARTITIONED_DAG",
+ schedule=CronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei"),
+ ):
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ data = {
+ "dag_id": "TEST_PARTITIONED_DAG",
+ "from_date": "2026-02-18T00:00:00+00:00",
+ "to_date": "2026-02-18T00:00:00+00:00",
+ "max_active_runs": 5,
+ "run_backwards": False,
+ }
+ response = test_client.post(url="/backfills", json=data)
+ assert response.status_code == 422
+
+ @mock.patch("airflow.api_fastapi.core_api.routes.public.backfills._do_dry_run", autospec=True)
+ def test_empty_window_dry_run_returns_200_with_zero_entries(
+ self, mock_dry_run, session, dag_maker, test_client
+ ):
+ """POST /backfills/dry_run with an empty window returns 200 with total_entries == 0."""
+ mock_dry_run.return_value = iter([])
+ with dag_maker(
+ session=session,
+ dag_id="TEST_PARTITIONED_DAG",
+ schedule=CronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei"),
+ ):
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ data = {
+ "dag_id": "TEST_PARTITIONED_DAG",
+ "from_date": "2026-02-18T00:00:00+00:00",
+ "to_date": "2026-02-18T00:00:00+00:00",
+ "max_active_runs": 5,
+ "run_backwards": False,
+ }
+ response = test_client.post(url="/backfills/dry_run", json=data)
+ assert response.status_code == 200
+ assert response.json()["total_entries"] == 0
+ assert response.json()["backfills"] == []
+
class TestCancelBackfill(TestBackfillEndpoint):
def test_cancel_backfill(self, session, test_client):
diff --git a/airflow-core/tests/unit/cli/commands/test_backfill_command.py b/airflow-core/tests/unit/cli/commands/test_backfill_command.py
index c897fa20443..5d5743349f7 100644
--- a/airflow-core/tests/unit/cli/commands/test_backfill_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_backfill_command.py
@@ -164,7 +164,8 @@ class TestCliBackfill:
from_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
to_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
reverse=reverse,
- reprocess_behavior="none",
+ reprocess_behavior=ReprocessBehavior.NONE,
+ dag_run_conf=None,
session=mock.ANY,
)
@@ -240,9 +241,10 @@ class TestCliBackfill:
with pytest.raises(SystemExit):
self.parser.parse_args(args)
+ @mock.patch("airflow.cli.commands.backfill_command.getuser", return_value="test_user")
@mock.patch("airflow.cli.commands.backfill_command._create_backfill")
- def test_backfill_with_empty_dag_run_conf(self, mock_create):
- """Test that empty dag_run_conf is properly parsed."""
+ def test_backfill_with_empty_dag_run_conf(self, mock_create, mock_getuser):
+ """Test that empty dag_run_conf ({}) is parsed as an empty dict, not None."""
args = [
"backfill",
"create",
@@ -265,6 +267,57 @@ class TestCliBackfill:
reverse=False,
dag_run_conf={},
reprocess_behavior=None,
- triggering_user_name="root",
+ triggering_user_name="test_user",
run_on_latest_version=True,
)
+
+ @mock.patch("airflow.cli.commands.backfill_command._do_dry_run", autospec=True)
+ def test_backfill_dry_run_passes_dag_run_conf(self, mock_dry_run):
+ """dry-run path forwards parsed dag_run_conf dict (not raw string) to _do_dry_run."""
+ mock_dry_run.return_value = iter([])
+ args = [
+ "backfill",
+ "create",
+ "--dag-id",
+ "example_bash_operator",
+ "--from-date",
+ DEFAULT_DATE.isoformat(),
+ "--to-date",
+ DEFAULT_DATE.isoformat(),
+ "--dry-run",
+ "--reprocess-behavior",
+ "failed",
+ "--dag-run-conf",
+ '{"key": "val"}',
+ ]
+ airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+
+ mock_dry_run.assert_called_once_with(
+ dag_id="example_bash_operator",
+ from_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
+ to_date=DEFAULT_DATE.replace(tzinfo=timezone.utc),
+ reverse=False,
+ reprocess_behavior=ReprocessBehavior.FAILED,
+ dag_run_conf={"key": "val"},
+ session=mock.ANY,
+ )
+
+ @mock.patch("airflow.cli.commands.backfill_command._create_backfill", autospec=True)
+ def test_backfill_empty_window_shows_friendly_message(self, mock_create):
+ """Empty-window backfill exits with code 1 and prints a message, no traceback."""
+ from airflow.models.backfill import NoBackfillRunsToCreate
+
+ mock_create.side_effect = NoBackfillRunsToCreate("No runs to create for Dag example_bash_operator")
+ args = [
+ "backfill",
+ "create",
+ "--dag-id",
+ "example_bash_operator",
+ "--from-date",
+ DEFAULT_DATE.isoformat(),
+ "--to-date",
+ DEFAULT_DATE.isoformat(),
+ ]
+ with pytest.raises(SystemExit) as exc_info:
+ airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+ assert exc_info.value.code == 1
diff --git a/airflow-core/tests/unit/models/test_backfill.py b/airflow-core/tests/unit/models/test_backfill.py
index 69a4ebd5dc2..bd1a4c1877c 100644
--- a/airflow-core/tests/unit/models/test_backfill.py
+++ b/airflow-core/tests/unit/models/test_backfill.py
@@ -20,6 +20,7 @@ from __future__ import annotations
from contextlib import nullcontext
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
+from unittest import mock
import pendulum
import pytest
@@ -37,6 +38,7 @@ from airflow.models.backfill import (
InvalidBackfillDateRange,
InvalidBackfillDirection,
InvalidReprocessBehavior,
+ NoBackfillRunsToCreate,
ReprocessBehavior,
_create_backfill,
_do_dry_run,
@@ -1384,6 +1386,94 @@ def test_partitioned_backfill_reprocess_failed(dag_maker, session):
assert bdr.partition_key == info.partition_key
+@mock.patch("airflow.models.backfill._get_info_list", autospec=True, return_value=[])
+def test_create_backfill_empty_window_raises_no_runs_to_create(mock_get_info_list, dag_maker, session):
+ """_create_backfill raises NoBackfillRunsToCreate when _get_info_list returns an empty list."""
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ with pytest.raises(NoBackfillRunsToCreate, match=dag.dag_id):
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2026-02-18"),
+ to_date=pendulum.parse("2026-02-18"),
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf={},
+ )
+
+
+@pytest.mark.parametrize("dag_run_conf", [None, {}])
+@mock.patch("airflow.models.backfill._get_info_list", autospec=True, return_value=[])
+def test_do_dry_run_empty_window_returns_empty_iterable(mock_get_info_list, dag_run_conf, dag_maker, session):
+ """_do_dry_run on an empty window yields nothing (does not raise), with or without dag_run_conf."""
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ infos = list(
+ _do_dry_run(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2026-02-18"),
+ to_date=pendulum.parse("2026-02-18"),
+ reverse=False,
+ reprocess_behavior=ReprocessBehavior.NONE,
+ dag_run_conf=dag_run_conf,
+ session=session,
+ )
+ )
+ assert infos == []
+
+
+def test_create_backfill_real_empty_window_no_orphan(dag_maker, session):
+ """_create_backfill with a real empty window raises NoBackfillRunsToCreate without leaving an orphan.
+
+ Uses a weekly-Monday timetable; 2026-02-18 is a Wednesday so _get_info_list returns [] for real.
+ After the raise, no incomplete Backfill row must exist — proving #1 (orphan fix) holds.
+ A second call for the same dag_id must succeed (not be blocked by AlreadyRunningBackfill).
+ """
+ with dag_maker(schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ session.commit()
+
+ wednesday = pendulum.datetime(2026, 2, 18, tz="UTC") # not a Monday
+
+ with pytest.raises(NoBackfillRunsToCreate, match=dag.dag_id):
+ _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=wednesday,
+ to_date=wednesday,
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ )
+
+ # No orphan: zero incomplete Backfill rows for this dag
+ orphan_count = session.scalar(
+ select(Backfill).where(
+ Backfill.dag_id == dag.dag_id,
+ Backfill.completed_at.is_(None),
+ )
+ )
+ assert orphan_count is None, "An orphan Backfill row was left behind after NoBackfillRunsToCreate"
+
+ # A valid (Monday) window must not be blocked by AlreadyRunningBackfill
+ monday = pendulum.datetime(2026, 2, 23, tz="UTC") # a Monday
+ br = _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=monday,
+ to_date=monday,
+ max_active_runs=2,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ )
+ assert br is not None
+
+
def test_handle_clear_run_preserves_partition_key(dag_maker, session):
"""BackfillDagRun created via the clear/reprocess path carries partition_key from info."""