Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion airflow-core/docs/authoring-and-scheduling/timetable.rst
Original file line number Diff line number Diff line change
Expand Up @@ -368,10 +368,14 @@ scheduled run, because the next run is advanced one period to avoid colliding
with the previous run's ``logical_date``. The reverse direction (data
interval -> trigger) does not skip a run.

This transition can happen without editing a Dag, in two ways:
This transition can happen without editing a Dag, in three ways:

- Flipping ``[scheduler] create_cron_data_intervals`` changes how every Dag
with a bare cron string in ``schedule=`` resolves its timetable.
- Flipping ``[scheduler] create_delta_data_intervals`` changes how every Dag
with a ``timedelta`` or ``relativedelta`` in ``schedule=`` resolves its timetable.
The same one-period skip applies when switching from ``DeltaTriggerTimetable``
to ``DeltaDataIntervalTimetable``.
- Crossing a version boundary where the default differs. Airflow 3 defaults
to ``False``; Airflow 2.x defaults to ``True``.

Expand Down
20 changes: 20 additions & 0 deletions airflow-core/docs/installation/upgrading_to_airflow3.rst
Original file line number Diff line number Diff line change
Expand Up @@ -388,6 +388,26 @@ These include:
Airflow 3 dagruns already exist (going
``CronTriggerTimetable`` -> ``CronDataIntervalTimetable``), one scheduled run
is skipped to avoid colliding with the previous run's ``logical_date``.

- The ``create_delta_data_intervals`` configuration is now **functional**. Previously, a bug caused
``create_cron_data_intervals`` to silently control timetable selection for both cron-string DAGs
**and** ``timedelta``/``relativedelta`` DAGs, making ``create_delta_data_intervals`` a no-op.

This affects users who set ``create_cron_data_intervals = True`` to preserve Airflow 2
data-interval semantics for cron DAGs (as the upgrade guide recommends). Because of the bug,
that flag was also keeping ``timedelta``/``relativedelta`` DAGs on ``DeltaDataIntervalTimetable``.
After this fix those DAGs fall through to ``DeltaTriggerTimetable`` instead, shifting
``logical_date``, ``ds``, ``ts``, and ``data_interval_*`` by one period. The same applies
to DAGs migrated via ``conversion_v1_to_v2``.

If you want ``timedelta``/``relativedelta`` DAGs to keep ``DeltaDataIntervalTimetable``
behavior, set ``create_delta_data_intervals = True`` explicitly.

Set this **before** the upgrade. Setting ``create_delta_data_intervals = True`` on the
current unpatched Airflow 3 is safe -- the key was previously ignored, so it takes effect
only once this fix is in place. If you flip this flag from ``False`` to ``True`` after
Airflow 3 DAG runs already exist (``DeltaTriggerTimetable`` -> ``DeltaDataIntervalTimetable``),
one scheduled run is skipped to avoid colliding with the previous run's ``logical_date``.
- **Manual Dag runs and data intervals**: In Airflow 3, do not assume that a manually triggered Dag run's ``data_interval`` is derived from, or equal to, the supplied ``logical_date``. If your Dag logic needs the user-specified trigger date, use ``logical_date`` explicitly. This especially affects workflows that read ``data_interval_start`` or ``data_interval_end`` during manual triggering or when using ``TriggerDagRunOperator``. For detailed migration guidance, see :ref:`data-interval-manual-triggering`.
- **Simple Auth** is now default ``auth_manager``. To continue using FAB as the Auth Manager, please install the FAB provider and set ``auth_manager`` to ``FabAuthManager``:

Expand Down
63 changes: 63 additions & 0 deletions airflow-core/newsfragments/69869.significant.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
Fix ``create_delta_data_intervals`` being ignored for timedelta/relativedelta schedules

``[scheduler] create_delta_data_intervals`` now correctly controls timetable selection
for DAGs that pass a ``timedelta`` or ``relativedelta`` to ``schedule=``. Before this
fix, ``create_cron_data_intervals`` silently governed both cron-string DAGs **and**
timedelta/relativedelta DAGs, making ``create_delta_data_intervals`` a no-op.

**Who is affected:**

Users who set ``create_cron_data_intervals = True`` in their Airflow 3 config to
preserve Airflow 2 data-interval semantics for cron DAGs (as recommended by
``upgrading_to_airflow3.rst``) are also unintentionally running their
``timedelta``/``relativedelta`` DAGs on ``DeltaDataIntervalTimetable``. After this
fix, those DAGs will fall through to the correct default (``DeltaTriggerTimetable``)
unless ``create_delta_data_intervals`` is also set to ``True``.

This also affects DAGs migrated from Airflow 2 via ``conversion_v1_to_v2``, which
rebuilds timetables through the same code path.

**Behaviour changes:**

- ``create_delta_data_intervals`` now takes effect as documented.
- ``create_cron_data_intervals`` now affects **only** cron-string DAGs.
- The two config keys are independent, as ``config.yml`` always documented.

**Migration:**

If you rely on ``DeltaDataIntervalTimetable`` for ``timedelta``/``relativedelta``
DAGs (i.e. you need contiguous data intervals and ``ds``/``ts`` anchored at
``data_interval_start``), set this **before** the next scheduler restart::

[scheduler]
create_delta_data_intervals = True

You can safely set this flag on current ``main`` (before this fix ships) because
``create_delta_data_intervals`` was previously a no-op - setting it to ``True``
will have no effect until the fix is in place, so there is no risk of an
unexpected timetable switch during the transition.

Switching a DAG from ``DeltaTriggerTimetable`` to ``DeltaDataIntervalTimetable``
(i.e. flipping ``create_delta_data_intervals`` from ``False`` to ``True`` when
existing DAG runs are present) skips one scheduled run to avoid colliding with
the previous run's ``logical_date``. Plan for this one-period gap or convert
affected DAGs to an explicit ``DeltaDataIntervalTimetable(...)`` instance in
``schedule=`` before the flag changes.

* Types of change

* [ ] DAG changes
* [x] Config changes
* [ ] API changes
* [ ] CLI changes
* [x] Behaviour changes
* [ ] Plugin changes
* [ ] Dependency changes
* [ ] Code interface changes

* Migration rules needed

* Users who set ``create_cron_data_intervals = True`` to preserve Airflow 2
data-interval behavior for cron DAGs should also set
``create_delta_data_intervals = True`` if they have ``timedelta``/
``relativedelta`` DAGs that depended on ``DeltaDataIntervalTimetable``.
4 changes: 4 additions & 0 deletions airflow-core/src/airflow/config_templates/config.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2836,6 +2836,10 @@ scheduler:
Notably, for **DeltaTriggerTimetable**, the logical date is the same as the time the DAG Run will
try to schedule, while for **DeltaDataIntervalTimetable**, 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 **DeltaTriggerTimetable** to **DeltaDataIntervalTimetable** (for example,
by flipping this setting from ``False`` to ``True``), the next scheduled run skips one period past
the most recent **DeltaTriggerTimetable** run to avoid colliding with its logical date.
version_added: 2.11.0
type: boolean
example: ~
Expand Down
2 changes: 1 addition & 1 deletion task-sdk/src/airflow/sdk/definitions/dag.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ def _create_timetable(interval: ScheduleInterval, timezone: Timezone | FixedTime
if interval == "@continuous":
return ContinuousTimetable()
if isinstance(interval, timedelta | relativedelta):
if airflow_conf.getboolean("scheduler", "create_cron_data_intervals"):
if airflow_conf.getboolean("scheduler", "create_delta_data_intervals"):
Comment thread
sharetheknowledge marked this conversation as resolved.
return DeltaDataIntervalTimetable(interval)
return DeltaTriggerTimetable(interval)
if isinstance(interval, str):
Expand Down
30 changes: 30 additions & 0 deletions task-sdk/tests/task_sdk/definitions/test_dag.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from unittest import mock

import pytest
from dateutil.relativedelta import relativedelta

from airflow.sdk import (
DAG,
Expand All @@ -38,9 +39,13 @@
from airflow.sdk.bases.operator import BaseOperator
from airflow.sdk.bases.timetable import BaseTimetable
from airflow.sdk.definitions.param import DagParam, ParamsDict
from airflow.sdk.definitions.timetables.interval import DeltaDataIntervalTimetable
from airflow.sdk.definitions.timetables.trigger import DeltaTriggerTimetable
from airflow.sdk.exceptions import AirflowDagCycleException, DuplicateTaskIdFound, RemovedInAirflow4Warning
from airflow.utils.types import DagRunType

from tests_common.test_utils.config import conf_vars

DEFAULT_DATE = datetime(2016, 1, 1, tzinfo=timezone.utc)


Expand Down Expand Up @@ -447,6 +452,31 @@ def test_continuous_schedule_linmits_max_active_runs(self):
with pytest.raises(ValueError, match="ContinuousTimetable requires max_active_runs <= 1"):
dag = DAG("continuous", start_date=DEFAULT_DATE, schedule="@continuous", max_active_runs=25)

@pytest.mark.parametrize("schedule", [timedelta(days=1), relativedelta(days=1)])
@pytest.mark.parametrize(
("delta", "cron", "expected"),
[
pytest.param("False", "False", DeltaTriggerTimetable, id="both-false"),
pytest.param("True", "False", DeltaDataIntervalTimetable, id="delta-true"),
pytest.param("False", "True", DeltaTriggerTimetable, id="cron-true"),
],
)
def test_timedelta_schedule_respects_create_delta_data_intervals_config(
self, schedule, delta, cron, expected
):
"""Regression guard: create_delta_data_intervals must control DeltaTriggerTimetable vs
DeltaDataIntervalTimetable for timedelta/relativedelta schedules, independently of
create_cron_data_intervals (which governs only cron-string schedules).
"""
with conf_vars(
{
("scheduler", "create_delta_data_intervals"): delta,
("scheduler", "create_cron_data_intervals"): cron,
}
):
dag = DAG("test_dag", start_date=DEFAULT_DATE, schedule=schedule)
assert isinstance(dag.timetable, expected)

def test_only_partitioned_at_runtime_has_partitioned_at_runtime_flag(self):
"""Regression guard: across every BaseTimetable subclass, only PartitionedAtRuntime sets partitioned_at_runtime=True."""

Expand Down