diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py index 5eb0ff40ff..84fafd0c19 100644 --- a/sqlmesh/core/scheduler.py +++ b/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 diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py index cd32d2451d..d7bb5e4497 100644 --- a/tests/core/test_scheduler.py +++ b/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", [