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
13 changes: 5 additions & 8 deletions sqlmesh/core/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
78 changes: 78 additions & 0 deletions tests/core/test_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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",
[
Expand Down