diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py index f2f432a97e..17b10c195a 100644 --- a/sqlmesh/core/plan/evaluator.py +++ b/sqlmesh/core/plan/evaluator.py @@ -34,6 +34,7 @@ SnapshotId, SnapshotInfoLike, SnapshotCreationFailedError, + missing_intervals_for_no_gaps, ) from sqlmesh.utils import to_snake_case from sqlmesh.core.state_sync import StateSync @@ -363,6 +364,44 @@ def visit_restatement_stage( def visit_environment_record_update_stage( self, stage: stages.EnvironmentRecordUpdateStage, plan: EvaluatablePlan ) -> None: + if ( + plan.no_gaps + and stage.no_gaps_snapshot_names + and not plan.is_dev + and not plan.skip_backfill + and not plan.empty_backfill + ): + # A plan's initial backfill frontier is frozen when the plan is built, but the + # target environment can advance before that plan reaches promotion. + target_environment = self.state_sync.get_environment(plan.environment.name) + if target_environment: + target_snapshots = list(stage.all_snapshots.values()) + self.state_sync.refresh_snapshot_intervals(target_snapshots) + + previous_snapshot_infos = [ + snapshot + for snapshot in target_environment.snapshots + if snapshot.name in stage.no_gaps_snapshot_names + ] + previous_snapshots = self.state_sync.get_snapshots(previous_snapshot_infos).values() + snapshot_to_intervals = missing_intervals_for_no_gaps( + target_snapshots, + previous_snapshots, + stage.no_gaps_snapshot_names, + ) + if snapshot_to_intervals: + self.visit_backfill_stage( + stages.BackfillStage( + snapshot_to_intervals=snapshot_to_intervals, + selected_snapshot_ids=stage.selected_snapshot_ids, + all_snapshots=stage.all_snapshots, + deployability_index=stage.deployability_index, + ), + plan, + ) + + # Promotion performs the same no-gaps check again, so a target that advances after + # the catch-up read still fails safely instead of being promoted with a gap. self.state_sync.promote( plan.environment, no_gaps_snapshot_names=stage.no_gaps_snapshot_names if plan.no_gaps else set(), diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py index 729e1705b4..1696a7be63 100644 --- a/sqlmesh/core/plan/stages.py +++ b/sqlmesh/core/plan/stages.py @@ -141,9 +141,15 @@ class EnvironmentRecordUpdateStage: Args: no_gaps_snapshot_names: Names of snapshots for which there should be no interval gaps. + all_snapshots: All snapshots in the plan by name. + selected_snapshot_ids: The snapshots to include in a just-in-time no-gaps backfill. + deployability_index: Deployability index for a just-in-time no-gaps backfill. """ no_gaps_snapshot_names: t.Set[str] + all_snapshots: t.Dict[str, Snapshot] + selected_snapshot_ids: t.Set[SnapshotId] + deployability_index: DeployabilityIndex @dataclass @@ -370,7 +376,10 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]: stages.append( EnvironmentRecordUpdateStage( - no_gaps_snapshot_names={s.name for s in before_promote_snapshots} + no_gaps_snapshot_names={s.name for s in before_promote_snapshots}, + all_snapshots=snapshots_by_name, + selected_snapshot_ids=before_promote_snapshots, + deployability_index=deployability_index, ) ) diff --git a/sqlmesh/core/snapshot/__init__.py b/sqlmesh/core/snapshot/__init__.py index 65e5c2a822..2d22450d45 100644 --- a/sqlmesh/core/snapshot/__init__.py +++ b/sqlmesh/core/snapshot/__init__.py @@ -24,6 +24,7 @@ has_paused_forward_only as has_paused_forward_only, merge_intervals as merge_intervals, missing_intervals as missing_intervals, + missing_intervals_for_no_gaps as missing_intervals_for_no_gaps, snapshots_to_dag as snapshots_to_dag, start_date as start_date, table_name as table_name, diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py index 0c9635a7c2..be2915c462 100644 --- a/sqlmesh/core/snapshot/definition.py +++ b/sqlmesh/core/snapshot/definition.py @@ -2314,6 +2314,38 @@ def start_date( return earliest +def missing_intervals_for_no_gaps( + target_snapshots: t.Iterable[Snapshot], + previous_snapshots: t.Iterable[Snapshot], + snapshot_names: t.Optional[t.Set[str]] = None, +) -> t.Dict[Snapshot, Intervals]: + """Find intervals target snapshots need to reach the previous snapshots' frontiers.""" + target_snapshots_by_name = {snapshot.name: snapshot for snapshot in target_snapshots} + cache: t.Dict[str, datetime] = {} + missing_intervals_by_snapshot: t.Dict[Snapshot, Intervals] = {} + + for previous_snapshot in previous_snapshots: + target_snapshot = target_snapshots_by_name.get(previous_snapshot.name) + if ( + target_snapshot is None + or target_snapshot.version == previous_snapshot.version + or (snapshot_names is not None and previous_snapshot.name not in snapshot_names) + or not target_snapshot.is_incremental + or not previous_snapshot.is_incremental + or not previous_snapshot.intervals + ): + continue + + start = to_timestamp(start_date(target_snapshot, target_snapshots_by_name.values(), cache)) + end = previous_snapshot.intervals[-1][1] + if start < end: + missing_intervals = target_snapshot.missing_intervals(start, end, end_bounded=True) + if missing_intervals: + missing_intervals_by_snapshot[target_snapshot] = missing_intervals + + return missing_intervals_by_snapshot + + def snapshots_to_dag(snapshots: t.Collection[Snapshot]) -> DAG[SnapshotId]: dag: DAG[SnapshotId] = DAG() for snapshot in snapshots: diff --git a/sqlmesh/core/state_sync/db/facade.py b/sqlmesh/core/state_sync/db/facade.py index 572e54b7f1..3adccd6533 100644 --- a/sqlmesh/core/state_sync/db/facade.py +++ b/sqlmesh/core/state_sync/db/facade.py @@ -20,7 +20,6 @@ import logging import typing as t from pathlib import Path -from datetime import datetime from sqlmesh.core.console import Console, get_console @@ -36,7 +35,7 @@ SnapshotIntervals, SnapshotNameVersion, SnapshotTableInfo, - start_date, + missing_intervals_for_no_gaps, ) from sqlmesh.core.snapshot.definition import ( Interval, @@ -62,7 +61,7 @@ from sqlmesh.core.state_sync.db.snapshot import SnapshotState from sqlmesh.core.state_sync.db.version import VersionState from sqlmesh.core.state_sync.db.migrator import StateMigrator, _backup_table_name -from sqlmesh.utils.date import TimeLike, to_timestamp, time_like_to_str, now_timestamp +from sqlmesh.utils.date import TimeLike, time_like_to_str, now_timestamp from sqlmesh.utils.errors import ConflictingPlanError, SQLMeshError logger = logging.getLogger(__name__) @@ -586,43 +585,25 @@ def _ensure_no_gaps( target_environment: Environment, snapshot_names: t.Optional[t.Set[str]], ) -> None: + target_snapshots = list(target_snapshots) target_snapshots_by_name = {s.name: s for s in target_snapshots} - - changed_version_prev_snapshots_by_name = { - s.name: s + changed_version_prev_snapshots = [ + s for s in target_environment.snapshots if s.name in target_snapshots_by_name and target_snapshots_by_name[s.name].version != s.version - } + ] + previous_snapshots = self.get_snapshots(changed_version_prev_snapshots).values() - prev_snapshots = self.get_snapshots( - changed_version_prev_snapshots_by_name.values() - ).values() - cache: t.Dict[str, datetime] = {} - - for prev_snapshot in prev_snapshots: - target_snapshot = target_snapshots_by_name[prev_snapshot.name] - if ( - (snapshot_names is None or prev_snapshot.name in snapshot_names) - and target_snapshot.is_incremental - and prev_snapshot.is_incremental - and prev_snapshot.intervals - ): - start = to_timestamp( - start_date(target_snapshot, target_snapshots_by_name.values(), cache) - ) - end = prev_snapshot.intervals[-1][1] - - if start < end: - missing_intervals = target_snapshot.missing_intervals( - start, end, end_bounded=True - ) - - if missing_intervals: - raise SQLMeshError( - f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " - "Please re-apply your plan to resolve this error." - ) + missing_intervals = missing_intervals_for_no_gaps( + target_snapshots, previous_snapshots, snapshot_names + ) + if missing_intervals: + target_snapshot = next(iter(missing_intervals)) + raise SQLMeshError( + f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " + "Please re-apply your plan to resolve this error." + ) @contextlib.contextmanager def _transaction(self) -> t.Iterator[None]: diff --git a/tests/core/test_plan_evaluator.py b/tests/core/test_plan_evaluator.py index 575f5ae742..0e44db49dd 100644 --- a/tests/core/test_plan_evaluator.py +++ b/tests/core/test_plan_evaluator.py @@ -1,4 +1,8 @@ +from datetime import date +from pathlib import Path + import pytest +import time_machine from pytest_mock.plugin import MockerFixture from sqlglot import parse_one @@ -82,3 +86,92 @@ def test_builtin_evaluator_push(sushi_context: Context, make_snapshot): ) assert sushi_context.engine_adapter.table_exists(new_model_snapshot.table_name()) assert sushi_context.engine_adapter.table_exists(new_view_model_snapshot.table_name()) + + +@pytest.mark.slow +@time_machine.travel("2026-08-06 01:00:00 UTC", tick=False) +def test_builtin_evaluator_catches_up_no_gaps_plan_to_live_prod_frontier( + tmp_path: Path, +) -> None: + def project_for(value: int) -> Path: + project_path = tmp_path / f"project_{value}" + models_path = project_path / "models" + models_path.mkdir(parents=True) + (project_path / "config.yaml").write_text( + f""" +default_gateway: local +gateways: + local: + connection: + type: duckdb + database: {tmp_path / "warehouse.db"} +model_defaults: + dialect: duckdb +""" + ) + (models_path / "daily.sql").write_text( + f""" +MODEL ( + name repro.daily, + kind INCREMENTAL_BY_TIME_RANGE ( + time_column ds, + lookback 1, + batch_size 1 + ), + cron '@daily', + start '2026-08-04' +); + +SELECT + @start_date AS ds, + {value} AS value +; +""" + ) + return project_path + + initial_project = project_for(1) + changed_project = project_for(2) + + initial_context = Context(paths=[initial_project]) + initial_plan = initial_context.plan_builder( + "prod", + no_gaps=True, + skip_tests=True, + skip_linter=True, + ).build() + initial_context.apply(initial_plan) + initial_context.close() + + context = Context(paths=[changed_project]) + try: + stale_plan = context.plan_builder( + "prod", + no_gaps=True, + skip_tests=True, + skip_linter=True, + ).build() + + with time_machine.travel("2026-08-07 01:00:00 UTC", tick=False): + context.run("prod") + assert context.engine_adapter.fetchall( + 'SELECT ds, value FROM "warehouse"."repro"."daily" ORDER BY ds' + ) == [ + (date(2026, 8, 4), 1), + (date(2026, 8, 5), 1), + (date(2026, 8, 6), 1), + ] + context.apply(stale_plan) + + prod_environment = context.state_sync.get_environment("prod") + assert prod_environment + assert prod_environment.plan_id == stale_plan.plan_id + assert context.engine_adapter.fetchall( + 'SELECT ds, value FROM "warehouse"."repro"."daily" ORDER BY ds' + ) == [ + (date(2026, 8, 4), 2), + (date(2026, 8, 5), 2), + (date(2026, 8, 6), 2), + ] + finally: + context.close()