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/23 05:54:53 UTC
(airflow) branch main updated: Fix partitioned backfill widening a sub-day window to the whole day (#68718)
This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 15b9666071f Fix partitioned backfill widening a sub-day window to the whole day (#68718)
15b9666071f is described below
commit 15b9666071f9a4dfc73f694d777ef0a203c7c479
Author: Wei Lee <we...@gmail.com>
AuthorDate: Tue Jun 23 13:54:46 2026 +0800
Fix partitioned backfill widening a sub-day window to the whole day (#68718)
---
.../src/airflow/serialization/definitions/dag.py | 27 ++---
airflow-core/src/airflow/timetables/base.py | 11 +-
airflow-core/src/airflow/timetables/trigger.py | 59 +++++++---
airflow-core/tests/unit/models/test_backfill.py | 130 +++++++++++++++------
airflow-core/tests/unit/models/test_dag.py | 41 ++++++-
.../unit/timetables/test_trigger_timetable.py | 92 +++++++++------
6 files changed, 253 insertions(+), 107 deletions(-)
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py
index 61205832ac5..9dc816eb9cc 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -470,20 +470,21 @@ class SerializedDAG:
earliest to latest.
For partitioned timetables, the iteration axis is ``partition_date``
- rather than ``logical_date``. ``earliest`` and ``latest`` are resolved
- to their respective day-bounds (timetable-timezone local midnight
- converted to UTC) and the iteration walks the half-open interval
- ``[resolve_day_bound(earliest.date()), resolve_day_bound(latest.date() + 1 day))``,
- making ``latest``'s calendar date inclusive regardless of its time
- component. Each yielded :class:`~airflow.timetables.base.DagRunInfo`
- has ``logical_date=None``, ``data_interval=None``, and
+ rather than ``logical_date``. The wall-clock reading of *earliest* and
+ *latest* (year, month, day, hour, minute, second) is re-interpreted in
+ the timetable's own timezone before alignment, so a UTC-midnight bound
+ supplied by the production backfill path is treated as the
+ timetable-local midnight — a cross-timezone daily backfill therefore
+ includes the first day rather than silently dropping it. The window
+ granularity follows the timetable's own partition cadence and is *not*
+ rounded up to whole calendar days, so a single-hour window on an hourly
+ timetable yields only the partitions inside it (08:00 and 09:00, not the
+ full day's 24). Each yielded
+ :class:`~airflow.timetables.base.DagRunInfo` has
+ ``logical_date=None``, ``data_interval=None``, and
``run_after == partition_date``; see
:meth:`~airflow.timetables.base.Timetable.iter_partition_dagrun_infos`
for details.
-
- Day-bound dispatch is centralised here rather than at every call site
- (e.g. backfill) so partitioned vs. non-partitioned routing is handled
- once. See https://github.com/apache/airflow/pull/67537#discussion_r3386682447
"""
if earliest is None:
earliest = self._time_restriction.earliest
@@ -494,8 +495,8 @@ class SerializedDAG:
if self.timetable.partitioned:
yield from self.timetable.iter_partition_dagrun_infos(
- earliest_date=earliest.date(),
- latest_date=latest.date(),
+ earliest=earliest,
+ latest=latest,
)
return
diff --git a/airflow-core/src/airflow/timetables/base.py b/airflow-core/src/airflow/timetables/base.py
index 400b63f3d70..34c8366d083 100644
--- a/airflow-core/src/airflow/timetables/base.py
+++ b/airflow-core/src/airflow/timetables/base.py
@@ -261,11 +261,16 @@ class Timetable(Protocol):
def iter_partition_dagrun_infos(
self,
*,
- earliest_date: datetime.date,
- latest_date: datetime.date,
+ earliest: datetime.datetime,
+ latest: datetime.datetime,
) -> Iterable[DagRunInfo]:
"""
- Yield one DagRunInfo per partition for calendar days in ``[earliest_date, latest_date]`` (both inclusive).
+ Yield one DagRunInfo per partition whose ``partition_date`` lies in ``[earliest, latest]`` (both inclusive).
+
+ The iteration granularity follows the timetable's own partition cadence
+ (e.g. one tick per hour for ``CronPartitionTimetable("0 * * * *")``), so a
+ sub-day window yields only the partitions inside it rather than every
+ partition of the surrounding calendar day.
Only called for partitioned timetables (``partitioned is True``). The default
implementation raises :exc:`NotImplementedError`; timetables that set
diff --git a/airflow-core/src/airflow/timetables/trigger.py b/airflow-core/src/airflow/timetables/trigger.py
index fdc9cbde91a..160085a8c4b 100644
--- a/airflow-core/src/airflow/timetables/trigger.py
+++ b/airflow-core/src/airflow/timetables/trigger.py
@@ -21,13 +21,18 @@ import functools
import math
import operator
import time
-from datetime import timedelta
from types import NoneType
from typing import TYPE_CHECKING, Any
import structlog
-from airflow._shared.timezones.timezone import coerce_datetime, make_aware, parse_timezone, utcnow
+from airflow._shared.timezones.timezone import (
+ coerce_datetime,
+ convert_to_utc,
+ make_aware,
+ parse_timezone,
+ utcnow,
+)
from airflow.exceptions import InvalidPartitionKeyError
from airflow.timetables._cron import CronMixin
from airflow.timetables._delta import DeltaMixin
@@ -463,17 +468,37 @@ class CronPartitionTimetable(CronTriggerTimetable):
partition_key = self._format_key(partition_date)
return partition_date, partition_key
+ def _localize_wall_clock_to_timetable_timezone(self, dt: datetime.datetime) -> DateTime:
+ """
+ Re-interpret *dt*'s wall-clock reading as a moment in this timetable's timezone.
+
+ The production backfill path attaches ``default_timezone`` (UTC) to the
+ bounds it passes in, but the user's intent is the same wall-clock time in
+ the timetable's local timezone. This method extracts the naive
+ year/month/day/hour/minute/second from *dt* (regardless of whatever
+ tzinfo it currently carries) and re-localises that wall-clock to
+ ``self._timezone``, returning a UTC instant.
+
+ Crucially it preserves sub-day precision — the hour, minute, and second
+ are kept intact — so a narrow hourly window is not widened to a whole day.
+ """
+ aware = coerce_datetime(dt)
+ naive = aware.replace(tzinfo=None)
+ return convert_to_utc(make_aware(naive, self._timezone))
+
def iter_partition_dagrun_infos(
self,
*,
- earliest_date: datetime.date,
- latest_date: datetime.date,
+ earliest: datetime.datetime,
+ latest: datetime.datetime,
) -> Iterable[DagRunInfo]:
"""
- Yield one DagRunInfo per cron tick for calendar days in ``[earliest_date, latest_date]`` (both inclusive).
+ Yield one DagRunInfo per cron tick whose partition_date lies in ``[earliest, latest]`` (both inclusive).
Iteration walks directly along the partition_date axis — one cron tick per
- partition — without any reverse mapping from run_after. Each tick yields:
+ partition — honoring the actual datetime window rather than rounding it to
+ whole calendar days, so a sub-day window (e.g. an hourly cron backfilled for
+ a single hour) yields only the ticks inside the window. Each tick yields:
- ``partition_date = current`` (the cron tick itself, as a UTC instant)
- ``partition_key`` formatted by :meth:`_format_key` (local-tz label)
@@ -489,16 +514,20 @@ class CronPartitionTimetable(CronTriggerTimetable):
final tiebreaker). Setting ``run_after = partition_date`` is the simplest
correct choice and avoids the need for a reverse mapping.
- :param earliest_date: inclusive lower bound calendar date; the UTC window start is
- ``resolve_day_bound(earliest_date)``, i.e. local midnight of that day.
- :param latest_date: inclusive upper bound calendar date; the UTC window end is
- ``resolve_day_bound(latest_date + 1 day)`` (exclusive), so all ticks within
- ``latest_date``'s local day are included.
+ :param earliest: inclusive lower bound on ``partition_date``; the wall-clock
+ reading of *earliest* is re-interpreted in the timetable's timezone before
+ alignment, so a UTC-midnight bound from the production backfill path is
+ treated as the timetable-local midnight rather than the UTC midnight.
+ Iteration starts at the first cron tick at or after that localized instant.
+ :param latest: inclusive upper bound on ``partition_date``; the same wall-clock
+ localization applies; the tick equal to the localized *latest* is included.
+
+ Both bounds must be timezone-aware; a naive datetime is coerced to UTC before
+ the wall-clock localization step.
"""
- earliest_partition_date = self.resolve_day_bound(earliest_date)
- latest_partition_date = self.resolve_day_bound(latest_date + timedelta(days=1))
- current = self._align_to_next(earliest_partition_date)
- while current < latest_partition_date:
+ current = self._align_to_next(self._localize_wall_clock_to_timetable_timezone(earliest))
+ latest_dt = self._localize_wall_clock_to_timetable_timezone(latest)
+ while current <= latest_dt:
partition_key = self._format_key(current)
yield DagRunInfo(
run_after=current,
diff --git a/airflow-core/tests/unit/models/test_backfill.py b/airflow-core/tests/unit/models/test_backfill.py
index 538c33a2461..69a4ebd5dc2 100644
--- a/airflow-core/tests/unit/models/test_backfill.py
+++ b/airflow-core/tests/unit/models/test_backfill.py
@@ -251,9 +251,14 @@ def test_create_backfill_partitioned(reverse, existing, start_date, dag_maker, s
PythonOperator(task_id="hi", python_callable=print)
# Map each partition_key label to the partition_date (UTC instant) the timetable computes.
+ # Bounds are passed tz-aware (timetable timezone) so they name the intended local partitions:
+ # partition_date is now honored as an instant, not rounded to a calendar day.
partition_date_by_key = {
info.partition_key: info.partition_date
- for info in dag.iter_dagrun_infos_between(pendulum.parse("2026-02-14"), pendulum.parse("2026-02-25"))
+ for info in dag.iter_dagrun_infos_between(
+ pendulum.datetime(2026, 2, 14, tz="Asia/Taipei"),
+ pendulum.datetime(2026, 2, 25, tz="Asia/Taipei"),
+ )
}
# Existing runs stand in for historical scheduled runs: they carry partition_date, which is
# what deduplication keys on (see _get_latest_dag_run_row_query).
@@ -271,8 +276,8 @@ def test_create_backfill_partitioned(reverse, existing, start_date, dag_maker, s
expected_run_conf = {"param1": "valABC"}
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-15"),
- to_date=pendulum.parse("2026-02-24"),
+ from_date=pendulum.datetime(2026, 2, 15, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 24, tz="Asia/Taipei"),
max_active_runs=2,
reverse=reverse,
triggering_user_name="pytest",
@@ -318,8 +323,8 @@ def test_create_backfill_partitioned_key_uses_timetable_timezone(dag_maker, sess
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-18"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
max_active_runs=1,
reverse=False,
triggering_user_name="pytest",
@@ -359,8 +364,8 @@ def test_backfill_partitioned_does_not_duplicate_legacy_utc_keyed_run(dag_maker,
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-18"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
max_active_runs=1,
reverse=False,
triggering_user_name="pytest",
@@ -986,8 +991,8 @@ def test_backfill_partitioned_with_partition_window(reverse, dag_maker, session)
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-20"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
max_active_runs=2,
reverse=reverse,
triggering_user_name="pytest",
@@ -1017,8 +1022,8 @@ def test_backfill_partitioned_at_cap_single_day(dag_maker, session):
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-18"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
max_active_runs=2,
reverse=False,
triggering_user_name="pytest",
@@ -1066,8 +1071,8 @@ def test_do_dry_run_with_partition_window(dag_maker, session):
infos = list(
_do_dry_run(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-20"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
reverse=False,
reprocess_behavior=ReprocessBehavior.NONE,
session=session,
@@ -1080,24 +1085,24 @@ def test_do_dry_run_with_partition_window(dag_maker, session):
@pytest.mark.db_test
def test_create_backfill_partitioned_non_utc_boundary(dag_maker, session):
"""
- Regression: partition_date selection axis must use calendar-date comparison in the
- timetable timezone, not a UTC-instant comparison.
-
- Asia/Taipei is UTC+8. A partition labelled "2026-02-15" has
- partition_date = 2026-02-14T16:00:00Z (the UTC instant of Taipei midnight).
- Passing that UTC instant through an instant comparison against the user's
- --partition-date-start 2026-02-15 (parsed as 2026-02-15T00:00:00Z) would
- yield 2026-02-14T16:00Z < 2026-02-15T00:00Z, dropping the boundary day.
- The date-label comparison avoids this by converting to timetable timezone
- before calling .date().
+ Guard: the production backfill path supplies UTC-midnight bounds; the first
+ partition-day must not be dropped even when the timetable timezone is UTC+8.
+
+ The CLI/API parses ``--from-date 2026-02-15`` as ``2026-02-15T00:00:00Z``
+ (UTC midnight) because ``default_timezone`` is UTC on a standard deployment.
+ Without wall-clock localization, ``_align_to_next`` would see this UTC instant and
+ return the first Taipei-midnight tick *after* UTC midnight — skipping the
+ 2026-02-15 partition (Taipei midnight = 2026-02-14T16:00Z) and starting from
+ 2026-02-15T16:00Z instead. The localization step re-interprets the UTC-midnight
+ wall-clock as a Taipei-midnight before alignment, so the first day is included.
"""
with dag_maker(schedule=CronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei")) as dag:
PythonOperator(task_id="hi", python_callable=print)
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-15"),
- to_date=pendulum.parse("2026-02-24"),
+ from_date=pendulum.datetime(2026, 2, 15, tz="UTC"),
+ to_date=pendulum.datetime(2026, 2, 24, tz="UTC"),
max_active_runs=10,
reverse=False,
triggering_user_name="pytest",
@@ -1128,6 +1133,55 @@ def test_create_backfill_partitioned_non_utc_boundary(dag_maker, session):
assert "2026-02-14" not in partition_date_labels
+@pytest.mark.db_test
+def test_create_backfill_partitioned_sub_day_window_not_widened(dag_maker, session):
+ """
+ Guard: a narrow hourly window must not be widened to a full day after the
+ wall-clock localization fix.
+
+ The production backfill path supplies UTC-midnight-style bounds. For an
+ hourly Taipei timetable (``0 * * * *``) the user's intent for a two-hour
+ window is exactly two partitions. If localization discarded sub-day precision
+ (e.g. by flooring to midnight), the result would be 24 partitions instead.
+
+ Window: 2026-02-15 08:00 UTC → 2026-02-15 09:00 UTC (production-path form).
+ Localized to Taipei these are 2026-02-15T08:00+08:00 and 2026-02-15T09:00+08:00,
+ i.e. UTC instants 2026-02-15T00:00Z and 2026-02-15T01:00Z — exactly two ticks.
+ """
+ with dag_maker(schedule=CronPartitionTimetable("0 * * * *", timezone="Asia/Taipei")) as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+
+ # Production-path bounds: UTC wall-clock (what CLI/API delivers on a UTC host).
+ # The user intended "2026-02-15 08:00 Taipei" and "2026-02-15 09:00 Taipei"
+ # — written as the corresponding UTC-wall-clock values a UTC machine produces.
+ from_date = pendulum.datetime(2026, 2, 15, 8, 0, 0, tz="UTC")
+ to_date = pendulum.datetime(2026, 2, 15, 9, 0, 0, tz="UTC")
+
+ b = _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=from_date,
+ to_date=to_date,
+ max_active_runs=10,
+ reverse=False,
+ triggering_user_name="pytest",
+ dag_run_conf=None,
+ )
+ query = (
+ select(DagRun)
+ .join(BackfillDagRun.dag_run)
+ .where(BackfillDagRun.backfill_id == b.id)
+ .order_by(BackfillDagRun.sort_ordinal)
+ )
+ dag_runs = session.scalars(query).all()
+ partition_hour_labels = [
+ pendulum.instance(x.partition_date).in_timezone("Asia/Taipei").strftime("%Y-%m-%dT%H:%M")
+ for x in dag_runs
+ ]
+
+ # Full-sequence equality: exactly the two requested partitions (in Taipei local time).
+ assert partition_hour_labels == ["2026-02-15T08:00", "2026-02-15T09:00"]
+
+
@pytest.mark.db_test
@pytest.mark.parametrize("run_offset", [1, 2, -1, -2])
def test_backfill_partitioned_nonzero_offset_full_window(run_offset, dag_maker, session):
@@ -1144,8 +1198,8 @@ def test_backfill_partitioned_nonzero_offset_full_window(run_offset, dag_maker,
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-20"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
max_active_runs=10,
reverse=False,
triggering_user_name="pytest",
@@ -1207,10 +1261,14 @@ def test_backfill_partitioned_nonzero_offset_at_cap_single_day(
PythonOperator(task_id="hi", python_callable=print)
session.commit()
+ def _parse_date_taipei(date_str: str) -> pendulum.DateTime:
+ y, m, d = (int(p) for p in date_str.split("-"))
+ return pendulum.datetime(y, m, d, tz="Asia/Taipei")
+
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse(from_date),
- to_date=pendulum.parse(to_date),
+ from_date=_parse_date_taipei(from_date),
+ to_date=_parse_date_taipei(to_date),
max_active_runs=2,
reverse=False,
triggering_user_name="pytest",
@@ -1242,8 +1300,8 @@ def test_backfill_partitioned_offset_zero_behavior_unchanged(dag_maker, session)
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-20"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
max_active_runs=5,
reverse=False,
triggering_user_name="pytest",
@@ -1274,7 +1332,11 @@ def test_partitioned_backfill_reprocess_failed(dag_maker, session):
# Determine the UTC partition_date for the 2026-02-18 Asia/Taipei partition.
info = next(
- i for i in dag.iter_dagrun_infos_between(pendulum.parse("2026-02-18"), pendulum.parse("2026-02-18"))
+ i
+ for i in dag.iter_dagrun_infos_between(
+ pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ )
)
expected_partition_date = info.partition_date
@@ -1292,8 +1354,8 @@ def test_partitioned_backfill_reprocess_failed(dag_maker, session):
b = _create_backfill(
dag_id=dag.dag_id,
- from_date=pendulum.parse("2026-02-18"),
- to_date=pendulum.parse("2026-02-18"),
+ from_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ to_date=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
max_active_runs=1,
reverse=False,
triggering_user_name="pytest",
diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py
index da751e7dd85..090cd2d4756 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -3438,8 +3438,8 @@ def test_iter_dagrun_infos_between_partitioned_timetable():
Verifies:
- Full-sequence equality: result matches direct iter_partition_dagrun_infos call.
- - to_date's calendar date is inclusive (half-open +1day upper bound makes it so).
- - to_date+1 is exclusive (partition tick for the day after to_date is absent).
+ - The datetime window bounds partition_date directly (no day rounding); both ends inclusive.
+ - to_date's partition is present; the tick after to_date is absent.
"""
dag = DAG(
dag_id="test_iter_dagrun_infos_between_partitioned",
@@ -3455,16 +3455,16 @@ def test_iter_dagrun_infos_between_partitioned_timetable():
result = list(scheduler_dag.iter_dagrun_infos_between(earliest=from_dt, latest=to_dt))
- # Build expected sequence by calling iter_partition_dagrun_infos directly with the same day-bounds.
+ # Build expected sequence by calling iter_partition_dagrun_infos directly with the same window.
core_timetable = scheduler_dag.timetable
expected = list(
core_timetable.iter_partition_dagrun_infos(
- earliest_date=from_dt.date(),
- latest_date=to_dt.date(),
+ earliest=from_dt,
+ latest=to_dt,
)
)
- # Full-sequence equality — not head/tail/length.
+ # Dispatch smoke-test: the partitioned path routes through iter_partition_dagrun_infos.
assert result == expected
# to_date's partition (2026-03-12) must be present (inclusive).
@@ -3476,6 +3476,35 @@ def test_iter_dagrun_infos_between_partitioned_timetable():
assert not any(info.run_after == after_to_date_partition for info in result)
+def test_iter_dagrun_infos_between_partitioned_subday_window_not_widened():
+ """Guard: a sub-day window through the full iter_dagrun_infos_between chain must
+ not widen to the whole calendar day.
+
+ Pre-fix the partitioned path truncated the bounds to calendar dates and expanded them to
+ whole UTC days, so an hourly Dag backfilled for a single hour produced ~24 runs instead of 2.
+ This drives the public chain (whose signature is unchanged), so on pre-fix code it yields the
+ widened 24-tick sequence; on the fixed code it yields exactly the two ticks in the window.
+ """
+ dag = DAG(
+ dag_id="test_iter_dagrun_infos_between_partitioned_subday",
+ start_date=DEFAULT_DATE,
+ schedule=CronPartitionTimetable("0 * * * *", timezone="UTC"),
+ )
+ EmptyOperator(task_id="dummy", dag=dag)
+ scheduler_dag = create_scheduler_dag(dag)
+
+ from_dt = pendulum.datetime(2026, 3, 10, 8, tz="UTC")
+ to_dt = pendulum.datetime(2026, 3, 10, 9, tz="UTC")
+
+ result = list(scheduler_dag.iter_dagrun_infos_between(earliest=from_dt, latest=to_dt))
+
+ # Exactly the two ticks inside [08:00, 09:00] — not the 24 ticks of the whole day.
+ assert [info.run_after for info in result] == [
+ pendulum.datetime(2026, 3, 10, 8, tz="UTC"),
+ pendulum.datetime(2026, 3, 10, 9, tz="UTC"),
+ ]
+
+
def test_iter_dagrun_infos_between_error(caplog):
start = pendulum.instance(DEFAULT_DATE - datetime.timedelta(hours=1))
end = pendulum.instance(DEFAULT_DATE)
diff --git a/airflow-core/tests/unit/timetables/test_trigger_timetable.py b/airflow-core/tests/unit/timetables/test_trigger_timetable.py
index 1806cc875e8..eef68b0679f 100644
--- a/airflow-core/tests/unit/timetables/test_trigger_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_trigger_timetable.py
@@ -918,15 +918,15 @@ def test_dagruninfo_backward_compatibility() -> None:
@pytest.mark.parametrize(
- ("run_offset", "earliest_date", "latest_date", "expected_triples"),
+ ("run_offset", "earliest", "latest", "expected_triples"),
[
pytest.param(
0,
# "0 0 * * *" UTC+8 (Asia/Taipei): ticks at 16:00 UTC each day.
- # earliest_date=2026-02-18 (local label) → UTC bound 2026-02-17T16:00Z
- # latest_date=2026-02-20 (inclusive) → UTC upper bound 2026-02-20T16:00Z
- datetime.date(2026, 2, 18),
- datetime.date(2026, 2, 20),
+ # earliest=2026-02-18 local midnight → 2026-02-17T16:00Z (first tick, inclusive)
+ # latest=2026-02-20 local midnight → 2026-02-19T16:00Z (last tick, inclusive)
+ pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
[
(
pendulum.datetime(2026, 2, 17, 16, tz="UTC"),
@@ -950,8 +950,8 @@ def test_dagruninfo_backward_compatibility() -> None:
1,
# offset=1: run_after = partition_date (same axis in new design)
# same three days as offset=0.
- datetime.date(2026, 2, 18),
- datetime.date(2026, 2, 20),
+ pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
[
(
pendulum.datetime(2026, 2, 17, 16, tz="UTC"),
@@ -974,8 +974,8 @@ def test_dagruninfo_backward_compatibility() -> None:
pytest.param(
-1,
# offset=-1: run_after = partition_date (same axis in new design)
- datetime.date(2026, 2, 18),
- datetime.date(2026, 2, 20),
+ pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
[
(
pendulum.datetime(2026, 2, 17, 16, tz="UTC"),
@@ -997,9 +997,7 @@ def test_dagruninfo_backward_compatibility() -> None:
),
],
)
-def test_iter_partition_dagrun_infos_full_sequence(
- run_offset, earliest_date, latest_date, expected_triples
-) -> None:
+def test_iter_partition_dagrun_infos_full_sequence(run_offset, earliest, latest, expected_triples) -> None:
"""iter_partition_dagrun_infos yields (partition_date, run_after, partition_key) full sequence.
Key behavioral invariant: run_after == partition_date for all offsets (including non-zero).
@@ -1007,7 +1005,7 @@ def test_iter_partition_dagrun_infos_full_sequence(
formats as the next calendar day's midnight.
"""
timetable = CoreCronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei", run_offset=run_offset)
- infos = list(timetable.iter_partition_dagrun_infos(earliest_date=earliest_date, latest_date=latest_date))
+ infos = list(timetable.iter_partition_dagrun_infos(earliest=earliest, latest=latest))
actual = [(info.partition_date, info.run_after, info.partition_key) for info in infos]
assert actual == expected_triples
# Invariants for every tick: run_after == partition_date (all offsets) and data_interval is None.
@@ -1020,29 +1018,29 @@ def test_iter_partition_dagrun_infos_full_sequence(
def test_iter_partition_dagrun_infos_empty_window() -> None:
- """Empty window (earliest_date > latest_date) yields an empty sequence."""
+ """Empty window (earliest > latest) yields an empty sequence."""
timetable = CoreCronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei", run_offset=0)
infos = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 2, 20),
- latest_date=datetime.date(2026, 2, 18),
+ earliest=pendulum.datetime(2026, 2, 20, tz="Asia/Taipei"),
+ latest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
)
)
assert infos == []
-def test_iter_partition_dagrun_infos_endpoint_not_on_tick() -> None:
- """When the UTC bound from resolve_day_bound is not on a cron tick, _align_to_next moves to the next tick.
+def test_iter_partition_dagrun_infos_lower_bound_not_on_tick() -> None:
+ """A lower bound that is not itself a cron tick is advanced to the next tick.
"0 6 * * *" Asia/Taipei ticks at 06:00 local = 22:00Z (previous UTC day).
- resolve_day_bound(2026-02-18) = 2026-02-17T16:00Z (local midnight), which is NOT a tick.
+ earliest = Taipei 2026-02-18 local midnight = 2026-02-17T16:00Z, which is NOT a tick.
_align_to_next(2026-02-17T16:00Z) moves to 2026-02-17T22:00Z (first 06:00 Taipei tick).
"""
timetable = CoreCronPartitionTimetable("0 6 * * *", timezone="Asia/Taipei", run_offset=0)
infos = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 2, 18),
- latest_date=datetime.date(2026, 2, 18),
+ earliest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ latest=pendulum.datetime(2026, 2, 18, 23, 59, 59, tz="Asia/Taipei"),
)
)
# Only one tick in the window: 2026-02-17T22:00Z (Taipei 2026-02-18T06:00).
@@ -1052,37 +1050,59 @@ def test_iter_partition_dagrun_infos_endpoint_not_on_tick() -> None:
def test_iter_partition_dagrun_infos_inclusive_endpoint_pair() -> None:
- """Both endpoints are inclusive calendar dates.
+ """Both endpoints are inclusive.
- latest_date=2026-02-18: window covers local day 2026-02-18 only (one tick).
- latest_date=2026-02-19: window extends to include 2026-02-19 (two ticks).
- The tick for the day after latest_date is always absent.
+ latest = the first tick: window covers one tick.
+ latest = the next day's tick: window covers two ticks.
+ The tick beyond latest is always absent.
"""
timetable = CoreCronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei", run_offset=0)
infos_one = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 2, 18),
- latest_date=datetime.date(2026, 2, 18),
+ earliest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ latest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
)
)
assert len(infos_one) == 1
assert infos_one[0].partition_key == "2026-02-18T00:00:00"
- # Extending latest_date by one calendar day adds exactly one more tick.
+ # Extending latest to the next day's tick adds exactly one more tick.
infos_two = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 2, 18),
- latest_date=datetime.date(2026, 2, 19),
+ earliest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ latest=pendulum.datetime(2026, 2, 19, tz="Asia/Taipei"),
)
)
assert len(infos_two) == 2
assert [i.partition_key for i in infos_two] == ["2026-02-18T00:00:00", "2026-02-19T00:00:00"]
- # The tick for 2026-02-20 (the day after latest_date) is NOT included.
+ # The tick for 2026-02-20 (beyond latest) is NOT included.
assert not any(i.partition_key == "2026-02-20T00:00:00" for i in infos_two)
+def test_iter_partition_dagrun_infos_subday_window_does_not_expand_to_whole_day() -> None:
+ """A sub-day window of an hourly timetable yields only the ticks inside it — not the whole day.
+
+ Guard: the partition-date backfill range used to round every window up to whole
+ calendar days, so backfilling an hourly Dag for a single hour produced ~24 runs (one per
+ hourly tick of that day) instead of the partitions actually requested.
+ """
+ timetable = CoreCronPartitionTimetable("0 * * * *", timezone="UTC", run_offset=0)
+ infos = list(
+ timetable.iter_partition_dagrun_infos(
+ earliest=pendulum.datetime(2026, 6, 18, 8, tz="UTC"),
+ latest=pendulum.datetime(2026, 6, 18, 9, tz="UTC"),
+ )
+ )
+ # The 08:00 and 09:00 ticks only (both endpoints inclusive), NOT 00:00..23:00.
+ assert [i.partition_key for i in infos] == ["2026-06-18T08:00:00", "2026-06-18T09:00:00"]
+ assert [i.partition_date for i in infos] == [
+ pendulum.datetime(2026, 6, 18, 8, tz="UTC"),
+ pendulum.datetime(2026, 6, 18, 9, tz="UTC"),
+ ]
+
+
def test_iter_partition_dagrun_infos_taipei_local_midnight_vs_utc_day() -> None:
"""Asia/Taipei midnight partitions are stored as the prior UTC day.
@@ -1092,11 +1112,11 @@ def test_iter_partition_dagrun_infos_taipei_local_midnight_vs_utc_day() -> None:
instant and the local key correctly reflect the timezone relationship.
"""
timetable = CoreCronPartitionTimetable("0 0 * * *", timezone="Asia/Taipei", run_offset=0)
- # Single calendar-day window for 2026-02-18 (local Taipei date).
+ # Single-partition window for the 2026-02-18 local Taipei midnight tick.
infos = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 2, 18),
- latest_date=datetime.date(2026, 2, 18),
+ earliest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
+ latest=pendulum.datetime(2026, 2, 18, tz="Asia/Taipei"),
)
)
assert len(infos) == 1
@@ -1118,8 +1138,8 @@ def test_iter_partition_dagrun_infos_dst_america_new_york_spring_forward() -> No
timetable = CoreCronPartitionTimetable("0 0 * * *", timezone="America/New_York", run_offset=0)
infos = list(
timetable.iter_partition_dagrun_infos(
- earliest_date=datetime.date(2026, 3, 7),
- latest_date=datetime.date(2026, 3, 8),
+ earliest=pendulum.datetime(2026, 3, 7, tz="America/New_York"),
+ latest=pendulum.datetime(2026, 3, 8, tz="America/New_York"),
)
)
partition_keys = [i.partition_key for i in infos]