From 4a756fcf905f8942817832258aaf86f01c2fbe7c Mon Sep 17 00:00:00 2001
From: sharetheknowledge <25279745+sharetheknowledge@users.noreply.github.com>
Date: Tue, 14 Jul 2026 13:33:00 +0200
Subject: [PATCH] Fix create_delta_data_intervals being ignored for timedelta
 schedules

_create_timetable() checked create_cron_data_intervals for both the
cron-string branch and the timedelta/relativedelta branch, so
create_delta_data_intervals had no effect and DAGs with a timedelta
schedule silently followed the cron config instead.

Closes #69868
---
 task-sdk/src/airflow/sdk/definitions/dag.py   |  2 +-
 .../tests/task_sdk/definitions/test_dag.py    | 40 +++++++++++++++++++
 2 files changed, 41 insertions(+), 1 deletion(-)

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..48494f570651c 100644
--- a/task-sdk/tests/task_sdk/definitions/test_dag.py
+++ b/task-sdk/tests/task_sdk/definitions/test_dag.py
@@ -447,6 +447,46 @@ 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)
 
+    def test_timedelta_schedule_respects_create_delta_data_intervals_config(self):
+        """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).
+        """
+        from airflow.sdk.definitions.timetables.interval import DeltaDataIntervalTimetable
+        from airflow.sdk.definitions.timetables.trigger import DeltaTriggerTimetable
+
+        from tests_common.test_utils.config import conf_vars
+
+        # Both False (the shipped airflow.cfg default): timedelta schedules use the trigger-based timetable.
+        with conf_vars(
+            {
+                ("scheduler", "create_delta_data_intervals"): "False",
+                ("scheduler", "create_cron_data_intervals"): "False",
+            }
+        ):
+            dag = DAG("delta_default", start_date=DEFAULT_DATE, schedule=timedelta(days=1))
+            assert isinstance(dag.timetable, DeltaTriggerTimetable)
+
+        # create_delta_data_intervals=True switches timedelta schedules to the data-interval timetable.
+        with conf_vars(
+            {
+                ("scheduler", "create_delta_data_intervals"): "True",
+                ("scheduler", "create_cron_data_intervals"): "False",
+            }
+        ):
+            dag = DAG("delta_data_interval", start_date=DEFAULT_DATE, schedule=timedelta(days=1))
+            assert isinstance(dag.timetable, DeltaDataIntervalTimetable)
+
+        # create_cron_data_intervals only affects cron/str schedules, not timedelta ones.
+        with conf_vars(
+            {
+                ("scheduler", "create_delta_data_intervals"): "False",
+                ("scheduler", "create_cron_data_intervals"): "True",
+            }
+        ):
+            dag = DAG("delta_unaffected_by_cron_config", start_date=DEFAULT_DATE, schedule=timedelta(days=1))
+            assert isinstance(dag.timetable, DeltaTriggerTimetable)
+
     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."""