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 6ccb4f6059a796ab530c82c4aafb5c2cff9187dd Author: Wei Lee <[email protected]> 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 [email protected]("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={}, + ) + + [email protected]("dag_run_conf", [None, {}]) [email protected]("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."""
