Fix scheduler stalling when a Dag switches to a coarser schedule

[TSKVenkat]

Changing a Dag from an hourly to a daily cron stops it being scheduled again. The scheduler keeps logging run already exists; skipping dagrun creation for a logical_date that is already taken and never advances past it, so the Dag silently stops producing runs until rows are deleted by hand.

_align_to_prev maps the previous run's interval end back onto the new schedule. When the new cron is coarser than the one that produced that run, the result is the boundary the run already occupies, so next_dagrun_info proposes a logical_date identical to the existing run's.

Reproduced on main (3.4.0) with the reporter's scenario — an hourly run at 2026-05-04 00:00 with interval [00:00, 01:00), then a switch to 0 0 * * *:

  previous run logical_date : 2026-05-04 00:00:00+00:00
  proposed next logical_date: 2026-05-04 00:00:00+00:00   <- collides, scheduler stalls

and after the change:

  proposed next logical_date: 2026-05-05 00:00:00+00:00   <- advances

Verified through the API the scheduler actually calls. SchedulerJobRunner reaches the timetable via next_dagrun_info_v2(), and the base Timetable.next_dagrun_info_v2 delegates to next_dagrun_info with last_dagrun_info.data_interval, so this is the live path rather than a helper reached only from tests:

last = DagRunInfo(
    run_after=pendulum.datetime(2026, 5, 4, 1, tz=UTC),
    data_interval=DataInterval(start=pendulum.datetime(2026, 5, 4, 0, tz=UTC),
                               end=pendulum.datetime(2026, 5, 4, 1, tz=UTC)),   # the hourly run
)
CronDataIntervalTimetable("0 0 * * *", UTC).next_dagrun_info_v2(          # now daily
    last_dagrun_info=last,
    restriction=TimeRestriction(earliest=..., latest=None, catchup=True),
)

At the merge base 82215cf97c:

  previous run logical_date : 2026-05-04 00:00:00+00:00
  proposed next logical_date: 2026-05-04 00:00:00+00:00   <- the run that already exists

With this change:

  proposed next logical_date: 2026-05-05 00:00:00+00:00

CronDataIntervalTimetable is reached when [scheduler] create_cron_data_intervals is true, which is the configuration the report names.

Scope of that check, stated plainly: this drives the scheduler's own timetable API, not a running scheduler against a metadata database. I could not stand a full scheduler up in this checkout — the Dag failed to deserialize in my local dev environment — so the last link, watching a dag_run row appear after the schedule change, is not something I have evidence for.

A guard for this already existed, added in #66132, but it only covered the zero-length intervals CronTriggerTimetable leaves behind (start == end). The collision is not specific to those: it happens whenever the previous run came from any finer schedule. Comparing the proposed logical_date against the previous run's expresses the invariant the scheduler actually depends on — a proposed run must be strictly later than the last one — and subsumes the zero-length case, so the existing test for it still passes unchanged.

Tests: test_coarser_schedule_does_not_re_emit_logical_date covers the hourly-to-daily switch for both catchup settings. It fails on main (2 failed) and passes with this change, while test_zero_length_last_interval_does_not_re_emit_logical_date passes on both sides. Full airflow-core/tests/unit/timetables/ run: 253 passed, 1 pre-existing failure in test_workday_timetable.py caused by pandas missing from my environment (it fails identically on unmodified main).

closes: #66754

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code following the guidelines

---

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.