Skip to content
Merged
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
3 changes: 3 additions & 0 deletions sqlmesh/core/state_sync/db/interval.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,9 @@ def remove_intervals(
for snapshot in all_snapshots
]

if not intervals_to_remove:
return

if logger.isEnabledFor(logging.INFO):
snapshot_ids = ", ".join(str(s.snapshot_id) for s, _ in intervals_to_remove)
logger.info("Removing interval for snapshots: %s", snapshot_ids)
Expand Down
38 changes: 38 additions & 0 deletions tests/core/integration/test_restatement.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,44 @@ def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
context.apply(plan)


@time_machine.travel("2023-01-08 15:00:00 UTC")
def test_prod_restatement_with_unbackfilled_dev_version(init_and_plan_context: t.Callable):
"""
Scenario:
Prod is built. A breaking change is planned to `dev` with `--skip-backfill`,
so `dev` holds a different snapshot version with no interval rows. Prod is
then restated.
Outcome:
RestatementStage tries to clear `dev` intervals for that other version, finds
none in `_intervals`, and no-ops instead of crashing on an empty insert. Prod
restatement still applies; the un-backfilled `dev` snapshot still has no
intervals.
"""
context, plan = init_and_plan_context("examples/sushi")
context.apply(plan)

prod_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id

context.upsert_model(
add_projection_to_model(t.cast(SqlModel, context.get_model("sushi.waiter_revenue_by_day")))
)
context.plan("dev", skip_backfill=True, auto_apply=True, no_prompts=True)

dev_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id
assert dev_snapshot_id != prod_snapshot_id
assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals

context.plan(
restate_models=["sushi.waiter_revenue_by_day"],
start="2023-01-07",
end="2023-01-08",
auto_apply=True,
no_prompts=True,
)

assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals


@time_machine.travel("2023-01-08 15:00:00 UTC")
def test_restatement_plan_across_environments_snapshot_with_shared_version(
init_and_plan_context: t.Callable,
Expand Down
34 changes: 34 additions & 0 deletions tests/core/state_sync/test_state_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,40 @@ def test_remove_interval_missing_snapshot(
]


def test_remove_interval_no_matching_intervals(
state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
) -> None:
snapshot = make_snapshot(
SqlModel(
name="a",
cron="@daily",
query=parse_one("select 1, ds"),
),
version="a",
)
state_sync.push_snapshots([snapshot])

# The snapshot has never been backfilled, so there are no rows to expand the shared versions from
state_sync.remove_intervals(
[(snapshot, snapshot.inclusive_exclusive("2020-01-15", "2020-01-17"))],
remove_shared_versions=True,
)

remove_records_count = state_sync.engine_adapter.fetchone(
"SELECT COUNT(*) FROM sqlmesh._intervals WHERE name = '\"a\"' AND version = 'a' AND is_removed"
)[0] # type: ignore
assert remove_records_count == 0

assert not state_sync.get_snapshots([snapshot])[snapshot.snapshot_id].intervals


def test_remove_interval_empty_input(state_sync: EngineAdapterStateSync) -> None:
state_sync.remove_intervals([])
state_sync.remove_intervals([], remove_shared_versions=True)

assert state_sync.engine_adapter.fetchone("SELECT COUNT(*) FROM sqlmesh._intervals")[0] == 0 # type: ignore


def test_refresh_snapshot_intervals(
state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
) -> None:
Expand Down
Loading