This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch backport-36a9a63ac0-v3-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ba0e9df5f1d5a813ec5f3a65352f6cb40c5aa5f9 Author: Sanghoon Kim / κΉμν <[email protected]> AuthorDate: Thu Sep 10 02:35:03 2026 +0900 Fix Calendar view computing cron planned runs in UTC instead of the Dag's timezone (#71243) * Fix Calendar planned runs using UTC instead of Dag timezone CalendarService._calculate_cron_planned_runs() handed croniter a UTC-tagged start_time, so cron fields were matched against UTC wall-clock instead of the timetable's configured timezone. Step planned instants with CronMixin._get_next() instead of iterating a raw croniter, so the calendar matches the scheduler exactly, including its DST gap/fold handling for every-hour crons. Closes #71234 * Add newsfragment for #71243 * Remove newsfragment; not needed for small bugfixes (cherry picked from commit 36a9a63ac08160772dd513605942eca45e5570ad) # Conflicts: # airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py --- .../api_fastapi/core_api/services/ui/calendar.py | 26 ++--- .../core_api/routes/ui/test_calendar.py | 121 ++++++++++++++++++++- 2 files changed, 132 insertions(+), 15 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py index 3c22472f734..373aca31088 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py @@ -17,14 +17,12 @@ from __future__ import annotations import collections -import itertools -from collections.abc import Iterator, Sequence +from collections.abc import Sequence from datetime import datetime from typing import Literal, cast import sqlalchemy as sa import structlog -from croniter.croniter import croniter from sqlalchemy.engine import Row from sqlalchemy.orm import InstrumentedAttribute, Session @@ -213,17 +211,17 @@ class CalendarService: """Calculate planned runs for cron-based timetables.""" dates: dict[datetime, int] = collections.Counter() - dates_iter: Iterator[datetime | None] = croniter( - cast("CronMixin", dag.timetable)._expression, - start_time=last_data_interval.end, - ret_type=datetime, - ) - - # Cap the iteration like _calculate_timetable_planned_runs does; a high-frequency - # expression (e.g. "* * * * *", or a seconds-resolution cron) would otherwise take - # hundreds of thousands of steps before hitting the year boundary. - for dt in itertools.islice(dates_iter, self.MAX_PLANNED_RUNS): - if dt is None or dt.year != year: + cron_timetable = cast("CronMixin", dag.timetable) + dt = last_data_interval.end + + # Step with CronMixin._get_next so planned instants match the scheduler exactly, + # including its DST gap/fold handling. Cap the iteration like + # _calculate_timetable_planned_runs does; a high-frequency expression (e.g. + # "* * * * *", or a seconds-resolution cron) would otherwise take hundreds of + # thousands of steps before hitting the year boundary. + for _ in range(self.MAX_PLANNED_RUNS): + dt = cron_timetable._get_next(dt) + if dt.year != year: break if dag.end_date and dt > dag.end_date: break diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py index 5c218e599c2..da07ae89108 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py @@ -25,7 +25,7 @@ from sqlalchemy.orm import Session from airflow.api_fastapi.core_api.services.ui.calendar import CalendarService from airflow.providers.standard.operators.empty import EmptyOperator -from airflow.sdk import CronPartitionTimetable +from airflow.sdk import CronPartitionTimetable, CronTriggerTimetable from airflow.utils.session import NEW_SESSION, provide_session from airflow.utils.state import DagRunState @@ -187,6 +187,125 @@ class TestCalendar: assert body == result +class TestCalendarCronNonUTCTimezone: + """Planned runs for a cron timetable must be computed in the timetable's own timezone, not UTC.""" + + DAG_NAME = "test_dag_non_utc_tz" + + @pytest.fixture(autouse=True) + @provide_session + def setup_dag_runs(self, dag_maker, *, session: Session = NEW_SESSION) -> None: + clear_db_runs() + clear_db_dags() + with dag_maker( + self.DAG_NAME, + schedule=CronTriggerTimetable("0 8 * * *", timezone="Asia/Seoul"), + start_date=datetime(2025, 1, 1), + catchup=True, + serialized=True, + session=session, + ): + EmptyOperator(task_id="test_task1") + dag_maker.create_dagrun( + run_id="run_1", + state=DagRunState.SUCCESS, + logical_date=pendulum.datetime(2025, 1, 1, 23, 0, 0, tz="UTC"), + ) + dag_maker.sync_dagbag_to_db() + + session.commit() + + def teardown_method(self) -> None: + clear_db_runs() + clear_db_dags() + + def test_planned_runs_use_timetable_timezone_not_utc(self, test_client): + response = test_client.get(f"/calendar/{self.DAG_NAME}", params={"granularity": "hourly"}) + assert response.status_code == 200 + body = response.json() + + planned = [r for r in body["dag_runs"] if r["state"] == "planned"] + # Daily 08:00 Asia/Seoul is 23:00Z the previous day; the last run's data interval + # ends 2025-01-01T23:00Z, so planned runs are one per remaining day of 2025. + assert len(planned) == 364 + assert min(r["date"] for r in planned) == "2025-01-02T23:00:00Z" + assert all(r["date"].endswith("T23:00:00Z") for r in planned), planned + assert all(r["count"] == 1 for r in planned) + + +class CalendarEveryHourCronDstBase: + """Every-hour crons must plan exactly one run per UTC hour across a DST transition, like the scheduler.""" + + DAG_NAME: str + START_DATE: datetime + LAST_RUN_UTC: pendulum.DateTime + EXPECTED_HOURS: list[str] + + @pytest.fixture(autouse=True) + @provide_session + def setup_dag_runs(self, dag_maker, *, session: Session = NEW_SESSION) -> None: + clear_db_runs() + clear_db_dags() + with dag_maker( + self.DAG_NAME, + schedule=CronTriggerTimetable("0 * * * *", timezone="Europe/Zurich"), + start_date=self.START_DATE, + catchup=True, + serialized=True, + session=session, + ): + EmptyOperator(task_id="test_task1") + dag_maker.create_dagrun( + run_id="run_1", + state=DagRunState.SUCCESS, + logical_date=self.LAST_RUN_UTC, + ) + dag_maker.sync_dagbag_to_db() + + session.commit() + + def teardown_method(self) -> None: + clear_db_runs() + clear_db_dags() + + def test_one_planned_run_per_utc_hour(self, test_client): + response = test_client.get(f"/calendar/{self.DAG_NAME}", params={"granularity": "hourly"}) + assert response.status_code == 200 + + planned = {r["date"]: r["count"] for r in response.json()["dag_runs"] if r["state"] == "planned"} + assert {h: planned.get(h) for h in self.EXPECTED_HOURS} == dict.fromkeys(self.EXPECTED_HOURS, 1) + + +class TestCalendarEveryHourCronDstFold(CalendarEveryHourCronDstBase): + """Fall-back (2025-10-26 03:00 CEST -> 02:00 CET): the repeated hour keeps its planned run.""" + + DAG_NAME = "test_dag_every_hour_dst_fold" + START_DATE = datetime(2025, 10, 1) + LAST_RUN_UTC = pendulum.datetime(2025, 10, 25, 22, 0, 0, tz="UTC") + EXPECTED_HOURS = [ + "2025-10-25T23:00:00Z", + "2025-10-26T00:00:00Z", + "2025-10-26T01:00:00Z", + "2025-10-26T02:00:00Z", + "2025-10-26T03:00:00Z", + ] + + +class TestCalendarEveryHourCronDstGap(CalendarEveryHourCronDstBase): + """Spring-forward (2026-03-29 02:00 CET -> 03:00 CEST): the skipped hour is not double-counted.""" + + DAG_NAME = "test_dag_every_hour_dst_gap" + START_DATE = datetime(2026, 3, 1) + LAST_RUN_UTC = pendulum.datetime(2026, 3, 28, 22, 0, 0, tz="UTC") + EXPECTED_HOURS = [ + "2026-03-28T23:00:00Z", + "2026-03-29T00:00:00Z", + "2026-03-29T01:00:00Z", + "2026-03-29T02:00:00Z", + "2026-03-29T03:00:00Z", + ] + + class TestPartitionedCalendar: """Calendar tests for partitioned Dags (AIP-76) which use partition_date instead of logical_date."""
