fix(scheduler): skip intervals that overlap unready parent intervals

`interval_diff` advanced past an unready interval whenever the
candidate interval started before it. When a daily model depends on an
hourly model whose signal marks an hour as not ready, the day containing
that hour was compared against the next unready interval instead, so the
daily model was evaluated over incomplete upstream data and the interval
was recorded as done. Ready intervals after the last overlap were also
dropped.

Always skip the candidate interval on overlap and keep every remaining
interval once the unready intervals are exhausted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: breken-ai <312387581+breken-ai@users.noreply.github.com>

Author: breken-ai

sqlmesh/core/scheduler.py

@@ -1158,15 +1158,12 @@ def interval_diff(
             if uninterrupted:
                 return results
 
-            if interval_a[0] >= interval_b[0]:
-                index_a += 1
-            else:
-                index_b += 1
+            # interval_a overlaps interval_b, so it must be excluded. interval_b is kept because
+            # it may also overlap the next interval in intervals_a.
+            index_a += 1
 
-    if index_a < len_a:
-        interval_a = intervals_a[index_a]
-        if interval_a[0] >= interval_b[1] or interval_b[0] >= interval_a[1]:
-            results.extend(intervals_a[index_a:])
+    # Any remaining intervals in intervals_a start after the last interval in intervals_b
+    results.extend(intervals_a[index_a:])
 
     return results
 

tests/core/test_scheduler.py

@@ -677,6 +677,15 @@ def test_interval_diff():
     assert interval_diff([(1, 2), (2, 3)], [(3, 4)], uninterrupted=True) == [(1, 2), (2, 3)]
     assert interval_diff([(1, 2), (2, 3)], [(2, 3)], uninterrupted=True) == [(1, 2)]
 
+    # An interval that contains a smaller interval from `intervals_b` (for example a daily
+    # interval that contains an unready hourly parent interval) must be excluded, and the
+    # intervals after it must still be kept.
+    assert interval_diff([(0, 24), (24, 48)], [(10, 11)]) == [(24, 48)]
+    assert interval_diff(
+        [(0, 24), (24, 48), (48, 72), (72, 96)],
+        [(10, 11), (53, 54)],
+    ) == [(24, 48), (72, 96)]
+
 
 def test_signal_intervals(mocker: MockerFixture, make_snapshot, get_batched_missing_intervals):
     @signal()
@@ -870,6 +879,75 @@ def signal_base(batch: DatetimeRanges):
     }
 
 
+def test_signals_unready_hourly_parent_blocks_daily_child(
+    mocker: MockerFixture, make_snapshot, get_batched_missing_intervals
+):
+    @signal()
+    def hourly_ready(batch: DatetimeRanges):
+        # 2023-01-01 10:00 and 2023-01-03 05:00 are not ready yet
+        return [i for i in batch if (i[0].day, i[0].hour) not in ((1, 10), (3, 5))]
+
+    signals = signal.get_registry()
+
+    parent = make_snapshot(
+        load_sql_based_model(
+            parse(  # type: ignore
+                """
+                MODEL (
+                    name hourly_parent,
+                    kind INCREMENTAL_BY_TIME_RANGE(
+                      time_column dt,
+                    ),
+                    cron '@hourly',
+                    start '2023-01-01',
+                    signals HOURLY_READY(),
+                );
+                SELECT @start_dt AS dt;
+                """
+            ),
+            signal_definitions=signals,
+        ),
+    )
+
+    child = make_snapshot(
+        load_sql_based_model(
+            parse(  # type: ignore
+                """
+                MODEL (
+                    name daily_child,
+                    kind INCREMENTAL_BY_TIME_RANGE(
+                      time_column dt,
+                    ),
+                    start '2023-01-01',
+                );
+                SELECT dt FROM hourly_parent
+                """
+            ),
+            signal_definitions=signals,
+        ),
+        nodes={parent.name: parent.model},
+    )
+
+    snapshot_evaluator = SnapshotEvaluator(adapters=mocker.MagicMock(), ddl_concurrent_tasks=1)
+    scheduler = Scheduler(
+        snapshots=[parent, child],
+        snapshot_evaluator=snapshot_evaluator,
+        state_sync=mocker.MagicMock(),
+        max_workers=2,
+        default_catalog=None,
+    )
+
+    batches = get_batched_missing_intervals(
+        scheduler, "2023-01-01", "2023-01-04", "2023-01-05 01:00:00"
+    )
+
+    # Days containing an unready parent hour must not be evaluated, the other days must be
+    assert batches[child] == [
+        (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
+        (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+    ]
+
+
 @pytest.mark.parametrize(
     "batch_size, expected_batches",
     [