This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 261fc1c9661 [v3-3-test] Fix drifting data intervals for monthly/yearly
schedules with catchup disabled (#69143) (#69189)
261fc1c9661 is described below
commit 261fc1c9661d046647b57f299818bf4d4c31d807
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Jul 1 14:30:05 2026 +0800
[v3-3-test] Fix drifting data intervals for monthly/yearly schedules with
catchup disabled (#69143) (#69189)
Co-authored-by: Shahar Epstein <[email protected]>
---
airflow-core/newsfragments/69143.bugfix.rst | 1 +
airflow-core/src/airflow/timetables/interval.py | 26 +++++--
.../unit/timetables/test_interval_timetable.py | 84 ++++++++++++++++++++++
3 files changed, 107 insertions(+), 4 deletions(-)
diff --git a/airflow-core/newsfragments/69143.bugfix.rst
b/airflow-core/newsfragments/69143.bugfix.rst
new file mode 100644
index 00000000000..f6a2b7880d9
--- /dev/null
+++ b/airflow-core/newsfragments/69143.bugfix.rst
@@ -0,0 +1 @@
+Fix drifting, mis-aligned data intervals for monthly and yearly
(``relativedelta``) schedules when ``catchup`` is disabled. The first interval
is now anchored to the Dag's ``start_date`` instead of a fixed 30-day/365-day
grid, matching the boundaries produced with ``catchup=True``.
diff --git a/airflow-core/src/airflow/timetables/interval.py
b/airflow-core/src/airflow/timetables/interval.py
index 5064ee9be50..30a7f5033fa 100644
--- a/airflow-core/src/airflow/timetables/interval.py
+++ b/airflow-core/src/airflow/timetables/interval.py
@@ -224,8 +224,26 @@ class DeltaDataIntervalTimetable(DeltaMixin,
_DataIntervalTimetable):
+ delta.seconds
)
- def _round(self, dt: DateTime) -> DateTime:
- """Round the given time to the nearest interval."""
+ def _round(self, dt: DateTime, anchor: DateTime) -> DateTime:
+ """
+ Floor ``dt`` to the latest schedule boundary at or before it.
+
+ Months/years have no fixed second count, so the epoch-grid rounding
+ used for fixed deltas would drift. For them we anchor on ``anchor``
+ (the start_date) and advance one period at a time, so boundaries match
+ the catchup=True grid -- relativedelta day-clamping is path-dependent
+ (e.g. Jan 31 -> Feb 28 -> Mar 28), so a multiplied jump would land
+ elsewhere. Fixed deltas keep the historical epoch rounding and ignore
+ ``anchor``.
+
+ ``anchor`` must be at or before ``dt``; otherwise the forward stepping
+ cannot reach ``dt`` and the result is meaningless.
+ """
+ if isinstance(self._delta, relativedelta) and (self._delta.months or
self._delta.years):
+ boundary = anchor
+ while self._get_next(boundary) <= dt:
+ boundary = self._get_next(boundary)
+ return boundary
if isinstance(self._delta, datetime.timedelta):
delta_in_seconds = self._delta.total_seconds()
else:
@@ -243,8 +261,8 @@ class DeltaDataIntervalTimetable(DeltaMixin,
_DataIntervalTimetable):
This is slightly different from the cron version at terminal values.
"""
- round_current_time = self._round(coerce_datetime(utcnow()))
- new_start = self._get_prev(round_current_time)
+ now = coerce_datetime(utcnow())
+ new_start = self._get_prev(self._round(now, earliest or now))
if earliest is None:
return new_start
return max(new_start, earliest)
diff --git a/airflow-core/tests/unit/timetables/test_interval_timetable.py
b/airflow-core/tests/unit/timetables/test_interval_timetable.py
index d8ad62131a8..dbe27c6784f 100644
--- a/airflow-core/tests/unit/timetables/test_interval_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_interval_timetable.py
@@ -43,6 +43,8 @@ OLD_INTERVAL = DataInterval(start=YESTERDAY, end=CURRENT_TIME)
HOURLY_CRON_TIMETABLE = CronDataIntervalTimetable("@hourly", utc)
HOURLY_TIMEDELTA_TIMETABLE =
DeltaDataIntervalTimetable(datetime.timedelta(hours=1))
HOURLY_RELATIVEDELTA_TIMETABLE =
DeltaDataIntervalTimetable(dateutil.relativedelta.relativedelta(hours=1))
+MONTHLY_RELATIVEDELTA_TIMETABLE =
DeltaDataIntervalTimetable(dateutil.relativedelta.relativedelta(months=1))
+YEARLY_RELATIVEDELTA_TIMETABLE =
DeltaDataIntervalTimetable(dateutil.relativedelta.relativedelta(years=1))
CRON_TIMETABLE = CronDataIntervalTimetable("30 16 * * *", utc)
DELTA_FROM_MIDNIGHT = datetime.timedelta(minutes=30, hours=16)
@@ -139,6 +141,88 @@ def test_no_catchup_next_info_starts_at_current_time(
assert next_info == DagRunInfo.interval(start=expected_start,
end=CURRENT_TIME)
[email protected](
+ "last_automated_data_interval",
+ [
+ pytest.param(None, id="first-run"),
+ pytest.param(
+ DataInterval(
+ pendulum.DateTime(2020, 1, 1, tzinfo=utc),
+ pendulum.DateTime(2020, 2, 1, tzinfo=utc),
+ ),
+ id="subsequent",
+ ),
+ ],
+)
[email protected](
+ ("timetable", "start_date", "expected_start", "expected_end"),
+ [
+ pytest.param(
+ MONTHLY_RELATIVEDELTA_TIMETABLE,
+ pendulum.DateTime(2025, 1, 15, tzinfo=utc),
+ pendulum.DateTime(2026, 5, 15, tzinfo=utc),
+ pendulum.DateTime(2026, 6, 15, tzinfo=utc),
+ id="monthly",
+ ),
+ pytest.param(
+ YEARLY_RELATIVEDELTA_TIMETABLE,
+ pendulum.DateTime(2020, 3, 10, tzinfo=utc),
+ pendulum.DateTime(2025, 3, 10, tzinfo=utc),
+ pendulum.DateTime(2026, 3, 10, tzinfo=utc),
+ id="yearly",
+ ),
+ ],
+)
+@time_machine.travel(pendulum.DateTime(2026, 6, 29, tzinfo=utc))
+def test_no_catchup_calendar_delta_aligns_to_start_date(
+ timetable: Timetable,
+ start_date: pendulum.DateTime,
+ expected_start: pendulum.DateTime,
+ expected_end: pendulum.DateTime,
+ last_automated_data_interval: DataInterval | None,
+) -> None:
+ """``catchup=False`` with a relativedelta in months/years must stay aligned
+ to ``start_date`` and not drift onto a fixed 30-day/365-day epoch grid."""
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=last_automated_data_interval,
+ restriction=TimeRestriction(earliest=start_date, latest=None,
catchup=False),
+ )
+ assert next_info == DagRunInfo.interval(start=expected_start,
end=expected_end)
+
+
+@time_machine.travel(pendulum.DateTime(2026, 6, 29, 12, tzinfo=utc))
+def test_no_catchup_calendar_delta_without_start_date_ends_now() -> None:
+ """With no ``start_date`` to anchor on, the interval simply ends at now."""
+ next_info = MONTHLY_RELATIVEDELTA_TIMETABLE.next_dagrun_info(
+ last_automated_data_interval=None,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=False),
+ )
+ assert next_info == DagRunInfo.interval(
+ start=pendulum.DateTime(2026, 5, 29, 12, tzinfo=utc),
+ end=pendulum.DateTime(2026, 6, 29, 12, tzinfo=utc),
+ )
+
+
+@time_machine.travel(pendulum.DateTime(2026, 6, 29, tzinfo=utc))
+def test_no_catchup_calendar_delta_uses_one_period_at_a_time_clamping() ->
None:
+ """Boundaries advance one relativedelta period at a time, so day-clamping
is
+ path-dependent: Jan 31 + 1 month clamps to Feb 28, and because each step
starts
+ from the previous boundary (not from Jan 31), the 28 then sticks --
+ Feb 28 -> Mar 28 -> ... -> May 28 -> Jun 28. A single multiplied jump off
+ ``start_date`` would instead re-clamp from the 31st and land elsewhere
+ (Jan 31 + 5 months -> Jun 30)."""
+ next_info = MONTHLY_RELATIVEDELTA_TIMETABLE.next_dagrun_info(
+ last_automated_data_interval=None,
+ restriction=TimeRestriction(
+ earliest=pendulum.DateTime(2025, 1, 31, tzinfo=utc), latest=None,
catchup=False
+ ),
+ )
+ assert next_info == DagRunInfo.interval(
+ start=pendulum.DateTime(2026, 5, 28, tzinfo=utc),
+ end=pendulum.DateTime(2026, 6, 28, tzinfo=utc),
+ )
+
+
@pytest.mark.parametrize(
"timetable",
[
