diff --git a/airflow-core/docs/authoring-and-scheduling/timetable.rst b/airflow-core/docs/authoring-and-scheduling/timetable.rst index b19cc80211392..907fee7ffdbb9 100644 --- a/airflow-core/docs/authoring-and-scheduling/timetable.rst +++ b/airflow-core/docs/authoring-and-scheduling/timetable.rst @@ -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``. diff --git a/airflow-core/docs/installation/upgrading_to_airflow3.rst b/airflow-core/docs/installation/upgrading_to_airflow3.rst index 91e86cb3ccb71..3721aaaf165f1 100644 --- a/airflow-core/docs/installation/upgrading_to_airflow3.rst +++ b/airflow-core/docs/installation/upgrading_to_airflow3.rst @@ -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``: diff --git a/airflow-core/newsfragments/69869.significant.rst b/airflow-core/newsfragments/69869.significant.rst new file mode 100644 index 0000000000000..d1bb3a29ef1bb --- /dev/null +++ b/airflow-core/newsfragments/69869.significant.rst @@ -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``. diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index af11f9fe701d1..e5fe5248b5cd1 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -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: ~ diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index 5ba0254b208b0..8207b1dbcdfc5 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -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"): return DeltaDataIntervalTimetable(interval) return DeltaTriggerTimetable(interval) if isinstance(interval, str): diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py b/task-sdk/tests/task_sdk/definitions/test_dag.py index 9b76816886c76..dfd22db6ef3f9 100644 --- a/task-sdk/tests/task_sdk/definitions/test_dag.py +++ b/task-sdk/tests/task_sdk/definitions/test_dag.py @@ -24,6 +24,7 @@ from unittest import mock import pytest +from dateutil.relativedelta import relativedelta from airflow.sdk import ( DAG, @@ -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) @@ -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."""