You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ur...@apache.org on 2026/05/02 10:35:16 UTC
(airflow) branch main updated: Don't re-emit logical_date when previous data_interval is zero-length (#66132)
This is an automated email from the ASF dual-hosted git repository.
uranusjr 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 4a9b5a0f70a Don't re-emit logical_date when previous data_interval is zero-length (#66132)
4a9b5a0f70a is described below
commit 4a9b5a0f70a79802ee1eac48a4580f926f00177f
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Sat May 2 04:35:07 2026 -0600
Don't re-emit logical_date when previous data_interval is zero-length (#66132)
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
Co-authored-by: Wei Lee <we...@gmail.com>
---
.../src/airflow/config_templates/config.yml | 4 ++++
airflow-core/src/airflow/timetables/interval.py | 14 ++++++++++++
airflow-core/tests/unit/models/test_dag.py | 13 +++++------
airflow-core/tests/unit/models/test_dagrun.py | 1 -
.../unit/timetables/test_interval_timetable.py | 25 ++++++++++++++++++++++
5 files changed, 50 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml
index 3ac011a0999..c37989b9b74 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -2624,6 +2624,10 @@ scheduler:
Notably, for **CronTriggerTimetable**, the logical date is the same as the time the DAG Run will
try to schedule, while for **CronDataIntervalTimetable**, the logical date is the beginning of
the data interval, but the DAG Run will try to schedule at the end of the data interval.
+
+ When a DAG is switched from **CronTriggerTimetable** to **CronDataIntervalTimetable** (for example,
+ by flipping this setting from ``False`` to ``True``), the next scheduled run skips one period past
+ the most recent **CronTriggerTimetable** run to avoid colliding with its logical date.
version_added: 2.9.0
type: boolean
example: ~
diff --git a/airflow-core/src/airflow/timetables/interval.py b/airflow-core/src/airflow/timetables/interval.py
index 1bc8d1067e3..5064ee9be50 100644
--- a/airflow-core/src/airflow/timetables/interval.py
+++ b/airflow-core/src/airflow/timetables/interval.py
@@ -110,6 +110,20 @@ class _DataIntervalTimetable(Timetable):
else:
# Data interval starts from the end of the previous interval.
start = align_last_data_interval_end
+
+ # CronTriggerTimetable stores its runs as point-in-time intervals
+ # (start == end == logical_date). After a switch to a
+ # CronDataIntervalTimetable the aligned `start` lands back on that
+ # same logical_date, so without this guard we'd propose a run
+ # identical to the existing one — which collides with the
+ # (dag_id, logical_date) unique constraint and leaves the scheduler
+ # looping on "run already exists; skipping dagrun creation" until
+ # the next period elapses. Advance one period to skip past it.
+ if (
+ last_automated_data_interval.start == last_automated_data_interval.end
+ and start == last_automated_data_interval.start
+ ):
+ start = self._get_next(start)
if restriction.latest is not None and start > restriction.latest:
return None
end = self._get_next(start)
diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py
index f6428423b71..4f437b44f51 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -623,7 +623,7 @@ class TestDag:
logical_date=model.next_dagrun,
run_type=DagRunType.SCHEDULED,
session=session,
- data_interval=(model.next_dagrun, model.next_dagrun),
+ data_interval=(model.next_dagrun, model.next_dagrun + timedelta(days=1)),
run_after=model.next_dagrun_create_after,
triggered_by=DagRunTriggeredByType.TEST,
)
@@ -638,7 +638,7 @@ class TestDag:
assert model.exceeds_max_non_backfill is True
if catchup is True:
- assert model.next_dagrun_create_after == DEFAULT_DATE + timedelta(days=1)
+ assert model.next_dagrun_create_after == DEFAULT_DATE + timedelta(days=2)
else:
assert model.next_dagrun_create_after > current_time + timedelta(days=-2) # allow for fuzz
@@ -1128,7 +1128,10 @@ class TestDag:
dag_run.execute_dag_callbacks(dag=dag, success=True)
@time_machine.travel(timezone.datetime(2025, 11, 11))
- @pytest.mark.parametrize(("catchup", "expected_next_dagrun"), [(True, DEFAULT_DATE), (False, None)])
+ @pytest.mark.parametrize(
+ ("catchup", "expected_next_dagrun"),
+ [(True, DEFAULT_DATE + datetime.timedelta(hours=1)), (False, None)],
+ )
def test_next_dagrun_after_fake_scheduled_previous(
self, catchup, expected_next_dagrun, testing_dag_bundle
):
@@ -1156,7 +1159,7 @@ class TestDag:
run_type=DagRunType.SCHEDULED,
logical_date=DEFAULT_DATE,
state=State.SUCCESS,
- data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE + delta),
)
sync_dag_to_db(dag)
with create_session() as session:
@@ -1169,8 +1172,6 @@ class TestDag:
# Verify next_dagrun_create_after is scheduled after next_dagrun
assert model.next_dagrun_create_after > model.next_dagrun
else:
- # For catchup=True, even though there is a run for this date already,
- # it is marked as manual/external, so we should create a scheduled one anyway!
assert model.next_dagrun == expected_next_dagrun
assert model.next_dagrun_create_after == expected_next_dagrun + delta
diff --git a/airflow-core/tests/unit/models/test_dagrun.py b/airflow-core/tests/unit/models/test_dagrun.py
index 5477e77d725..34e8acbaacf 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -2264,7 +2264,6 @@ def test_schedule_tis_only_one_scheduler_update_succeeds_when_competing(dag_make
assert refreshed_ti.try_number == 1
-@pytest.mark.xfail(reason="We can't keep this behaviour with remote workers where scheduler can't reach xcom")
@pytest.mark.need_serialized_dag
def test_schedule_tis_start_trigger(dag_maker, session):
"""
diff --git a/airflow-core/tests/unit/timetables/test_interval_timetable.py b/airflow-core/tests/unit/timetables/test_interval_timetable.py
index 5fc836e23fd..d8ad62131a8 100644
--- a/airflow-core/tests/unit/timetables/test_interval_timetable.py
+++ b/airflow-core/tests/unit/timetables/test_interval_timetable.py
@@ -65,6 +65,31 @@ def test_no_catchup_first_starts_at_current_time(
assert next_info == DagRunInfo.interval(start=expected_start, end=CURRENT_TIME + DELTA_FROM_MIDNIGHT)
+@pytest.mark.parametrize(
+ "catchup",
+ [pytest.param(True, id="catchup_true"), pytest.param(False, id="catchup_false")],
+)
+@time_machine.travel(pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc))
+def test_zero_length_last_interval_does_not_re_emit_logical_date(catchup: bool) -> None:
+ """A zero-length ``data_interval`` (``start == end``) on the previous run
+ must not cause ``next_dagrun_info`` to re-emit that run's logical_date.
+
+ These appear when a DAG was scheduled by ``CronTriggerTimetable`` and later
+ switched to ``CronDataIntervalTimetable``. Without the guard the scheduler
+ loops on "run already exists; skipping dagrun creation".
+ """
+ timetable = CronDataIntervalTimetable("0 17 * * *", utc)
+ last_run_at = pendulum.DateTime(2021, 9, 5, 17, tzinfo=utc)
+ last = DataInterval(start=last_run_at, end=last_run_at)
+ next_info = timetable.next_dagrun_info(
+ last_automated_data_interval=last,
+ restriction=TimeRestriction(earliest=None, latest=None, catchup=catchup),
+ )
+ expected_start = pendulum.DateTime(2021, 9, 6, 17, tzinfo=utc)
+ expected_end = pendulum.DateTime(2021, 9, 7, 17, tzinfo=utc)
+ assert next_info == DagRunInfo.interval(start=expected_start, end=expected_end)
+
+
@pytest.mark.parametrize(
"earliest",
[pytest.param(None, id="none"), pytest.param(START_DATE, id="start_date")],