From bfa224b0a998d69627538bde30be523a843e6a48 Mon Sep 17 00:00:00 2001 From: Jesse Hodges Date: Wed, 21 Oct 2026 16:50:00 -0500 Subject: [PATCH 1/3] feat: allow bounded trailing gaps during production promotion Signed-off-by: Jesse Hodges --- docs/integrations/github.md | 45 ++ sqlmesh/core/context.py | 68 ++- sqlmesh/core/plan/builder.py | 10 + sqlmesh/core/plan/definition.py | 9 + sqlmesh/core/plan/evaluator.py | 15 +- sqlmesh/core/plan/stages.py | 14 +- sqlmesh/core/promotion.py | 339 +++++++++++++ sqlmesh/core/state_sync/base.py | 9 + sqlmesh/core/state_sync/db/facade.py | 164 +++++- sqlmesh/integrations/github/cicd/config.py | 1 + .../integrations/github/cicd/controller.py | 1 + tests/core/state_sync/test_state_sync.py | 465 ++++++++++++++++- tests/core/test_plan_policy.py | 384 ++++++++++++++ tests/core/test_promotion.py | 480 ++++++++++++++++++ tests/core/test_trailing_gap_integration.py | 224 ++++++++ tests/integrations/github/cicd/test_config.py | 13 +- .../github/cicd/test_github_controller.py | 39 ++ 17 files changed, 2262 insertions(+), 18 deletions(-) create mode 100644 sqlmesh/core/promotion.py create mode 100644 tests/core/test_plan_policy.py create mode 100644 tests/core/test_promotion.py create mode 100644 tests/core/test_trailing_gap_integration.py diff --git a/docs/integrations/github.md b/docs/integrations/github.md index 328acd0d17..c9714e64b2 100644 --- a/docs/integrations/github.md +++ b/docs/integrations/github.md @@ -300,11 +300,56 @@ Below is an example of how to define the default config for the bot in either YA | `skip_pr_backfill` | Indicates if the bot should skip backfilling models in the PR environment. Default: `True` | bool | N | | `pr_include_unmodified` | Indicates whether to include unmodified models in the PR environment. Default to the project's config value (which defaults to `False`) | bool | N | | `run_on_deploy_to_prod` | Indicates whether to run latest intervals when deploying to prod. If set to false, the deployment will backfill only the changed models up to the existing latest interval in production, ignoring any missing intervals beyond this point. Default: `False` | bool | N | +| `max_trailing_gap_intervals_on_deploy_to_prod` | Maximum number of scheduler boundaries of newly observed trailing gaps that a production deployment may accept. `0` keeps the strict no-gap behavior (default); a positive value opts in to bounded trailing-gap promotion. | int | N | | `pr_environment_name` | The name of the PR environment to create for which a PR number will be appended to. Defaults to the repo name if not provided. Note: The name will be normalized to alphanumeric + underscore and lowercase. | str | N | | `prod_branch_name` | The name of the git branch associated with production. Ex: `prod`. Default: `main` or `master` is considered prod | str | N | | `forward_only_branch_suffix` | If the git branch has this suffix, trigger a [forward-only](../concepts/plans.md#forward-only-plans) plan instead of a normal plan. Default: `-forward-only` | str | N | | `check_if_blocked_on_deploy_to_prod` | The bot normally checks if a PR is blocked from merging before deploying to production. Setting this to `False` will skip that check. Default: `True` | bool | N | +### Bounded trailing gaps during production promotion + +`max_trailing_gap_intervals_on_deploy_to_prod` handles a scheduler race while a +production plan is being built. It defaults to `0`, which preserves strict +no-gap promotion. Set it to a finite, +nonnegative integer to allow a production plan with `no_gaps=True` to accept a +bounded, contiguous trailing suffix that is observed after the plan's immutable +plan-time frontier. The candidate must still contain complete coverage through +that frontier. Historical, internal, ambiguous, and over-bound gaps continue to +block promotion. + +As a conservative first-release guard, promotion also rejects an otherwise +eligible suffix when any transitive downstream model already has recorded +coverage overlapping that suffix. A normal scheduler repair can fill the +upstream gap, but it does not automatically invalidate downstream intervals +that are already recorded, so the deployment fails closed in this case. + +The bound counts actual scheduler boundaries in stored interval coverage. It +does not count lookback-expanded work, so one accepted missing interval can +cause Cron to plan more intervals when it repairs a model with lookback. After +promotion, the normal Cron scheduler discovers the missing suffix and repairs +it using the usual dependency ordering and lookback behavior. Any subsequent +scheduler selection must include the affected models' upstream dependencies (or +use the scheduler's normal automatic upstream selection). A downstream-only +selection can record incomplete data before its upstream models are repaired; +this option does not automatically invalidate that work. + +The option is passed only to the production `no_gaps` plan. PR, development, +test, and production plans built with `no_gaps=False` retain their existing +behavior. The promotion warning identifies the environment, plan, snapshots, +frontiers, and accepted range so the temporary freshness debt is observable. +The warning reports the observed debt at classification time. SQLMesh does not +fence a scheduler that is already in flight, so additional intervals may be +recorded on the incumbent after that check; the next Cron run must reconcile +any remaining coverage on the promoted candidate. + +For example: + +```yaml +cicd_bot: + type: github + max_trailing_gap_intervals_on_deploy_to_prod: 1 +``` + Example with all properties defined: === "YAML" diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index 39590a0fa3..b06fd90b33 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -91,6 +91,7 @@ ) from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals, PlanExplainer from sqlmesh.core.plan.definition import UserProvidedFlags +from sqlmesh.core.promotion import PromotionFrontier from sqlmesh.core.reference import ReferenceGraph from sqlmesh.core.scheduler import Scheduler, CompletionStatus from sqlmesh.core.schema_loader import create_external_models_file @@ -1461,6 +1462,7 @@ def plan( explain: t.Optional[bool] = None, ignore_cron: t.Optional[bool] = None, min_intervals: t.Optional[int] = None, + max_trailing_gap_intervals: t.Optional[int] = None, ) -> Plan: """Interactively creates a plan. @@ -1511,6 +1513,8 @@ def plan( explain: Whether to explain the plan instead of applying it. min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered on every model when checking for missing intervals + max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that production + promotion may accept as debt when no_gaps is enabled. Returns: The populated Plan object. @@ -1543,6 +1547,7 @@ def plan( explain=explain, ignore_cron=ignore_cron, min_intervals=min_intervals, + max_trailing_gap_intervals=max_trailing_gap_intervals, ) plan = plan_builder.build() @@ -1601,6 +1606,7 @@ def plan_builder( ignore_cron: t.Optional[bool] = None, min_intervals: t.Optional[int] = None, always_include_local_changes: t.Optional[bool] = None, + max_trailing_gap_intervals: t.Optional[int] = None, ) -> PlanBuilder: """Creates a plan builder. @@ -1644,6 +1650,8 @@ def plan_builder( on every model when checking for missing intervals always_include_local_changes: Usually when restatements are present, local changes in the filesystem are ignored. However, it can be desirable to deploy changes + restatements in the same plan, so this flag overrides the default behaviour. + max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that production + promotion may accept as debt when no_gaps is enabled. Returns: The plan builder. @@ -1696,6 +1704,9 @@ def plan_builder( diff_rendered = diff_rendered or False skip_linter = skip_linter or False min_intervals = min_intervals or 0 + max_trailing_gap_intervals = ( + 0 if max_trailing_gap_intervals is None else max_trailing_gap_intervals + ) environment = environment or self.config.default_target_environment environment = Environment.sanitize_name(environment) @@ -1821,12 +1832,29 @@ def plan_builder( backfill_models = modified_model_names or None max_interval_end_per_model = None + planned_frontiers: t.Dict[str, PromotionFrontier] = {} default_start, default_end = None, None - if not run: - ignore_cron = False - max_interval_end_per_model = self._get_max_interval_end_per_model( + should_capture_frontiers = not is_dev and no_gaps and max_trailing_gap_intervals > 0 + if not run or should_capture_frontiers: + captured_max_interval_end_per_model = self._get_max_interval_end_per_model( snapshots, backfill_models ) + if should_capture_frontiers: + planned_frontiers = self._get_planned_frontiers( + context_diff, captured_max_interval_end_per_model + ) + + if run: + # `--run` deliberately does not cap backfills at the current production + # frontier. Keep the captured values separate so recording promotion + # frontiers does not change that behavior. + max_interval_end_per_model = None + else: + max_interval_end_per_model = captured_max_interval_end_per_model + + if not run: + ignore_cron = False + assert max_interval_end_per_model is not None # If no end date is specified, use the max interval end from prod # to prevent unintended evaluation of the entire DAG. default_start, default_end = self._get_plan_default_start_end( @@ -1925,6 +1953,8 @@ def plan_builder( }, explain=explain or False, ignore_cron=ignore_cron or False, + max_trailing_gap_intervals=max_trailing_gap_intervals, + planned_frontiers=planned_frontiers, ) def apply( @@ -3588,6 +3618,38 @@ def _get_max_interval_end_per_model( ).items() } + @staticmethod + def _get_planned_frontiers( + context_diff: ContextDiff, + max_interval_end_per_model: t.Mapping[str, datetime], + ) -> t.Dict[str, PromotionFrontier]: + """Capture production frontiers for changed models at plan-build time. + + The interval-end query is intentionally performed before any execution-time or + explicit-end adjustment. The incumbent identity comes from the context diff's + modified snapshot tuple and is carried with the frontier so promotion can verify + it against the incumbent observed under its environment-row lock. A missing + previous plan means there is no incumbent identity to carry, so no frontier is + recorded and promotion remains strict. + """ + if not context_diff.previous_plan_id: + return {} + + frontiers: t.Dict[str, PromotionFrontier] = {} + + for name, (current_snapshot, previous_snapshot) in context_diff.modified_snapshots.items(): + if not current_snapshot.is_incremental or not previous_snapshot.is_incremental: + continue + if name not in max_interval_end_per_model: + continue + + frontiers[name] = PromotionFrontier( + snapshot_id=previous_snapshot.snapshot_id, + end=to_timestamp(max_interval_end_per_model[name]), + ) + + return frontiers + @staticmethod def _get_models_for_interval_end( snapshots: t.Dict[str, Snapshot], backfill_models: t.Set[str] diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index a6307a9ffd..034b1ed373 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -23,6 +23,7 @@ UserProvidedFlags, earliest_interval_start, ) +from sqlmesh.core.promotion import PromotionFrontier from sqlmesh.core.schema_diff import ( get_schema_differ, has_drop_alteration, @@ -72,6 +73,9 @@ class PlanBuilder: no_gaps: Whether to ensure that new snapshots for nodes that are already a part of the target environment have no data gaps when compared against previous snapshots for same nodes. + max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that + production promotion may accept as debt when no_gaps is enabled. + planned_frontiers: The production snapshot frontiers captured when the plan was built. skip_backfill: Whether to skip the backfill step. empty_backfill: Like skip_backfill, but also records processed intervals. is_dev: Whether this plan is for development purposes. @@ -138,6 +142,8 @@ def __init__( console: t.Optional[PlanBuilderConsole] = None, user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None, selected_models: t.Optional[t.Set[str]] = None, + max_trailing_gap_intervals: int = 0, + planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None, ): self._context_diff = context_diff self._no_gaps = no_gaps @@ -184,6 +190,8 @@ def __init__( self._user_provided_flags = user_provided_flags self._selected_models = selected_models self._explain = explain + self._max_trailing_gap_intervals = max_trailing_gap_intervals + self._planned_frontiers = dict(planned_frontiers or {}) self._start = start if not self._start and self._forward_only_preview_needed: @@ -375,6 +383,8 @@ def build(self) -> Plan: ignore_cron=self._ignore_cron, user_provided_flags=self._user_provided_flags, selected_models=self._selected_models, + max_trailing_gap_intervals=self._max_trailing_gap_intervals, + planned_frontiers=self._planned_frontiers, ) self._latest_plan = plan return plan diff --git a/sqlmesh/core/plan/definition.py b/sqlmesh/core/plan/definition.py index 866299eff8..d72dde3612 100644 --- a/sqlmesh/core/plan/definition.py +++ b/sqlmesh/core/plan/definition.py @@ -9,6 +9,7 @@ from sqlmesh.core.context_diff import ContextDiff from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements +from sqlmesh.core.promotion import PromotionFrontier from sqlmesh.utils.metaprogramming import Executable # noqa from sqlmesh.core.node import IntervalUnit from sqlmesh.core.snapshot import ( @@ -83,6 +84,10 @@ class Plan(PydanticModel, frozen=True): user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None selected_models: t.Optional[t.Set[str]] = None """Models that have been selected for this plan (used for dbt selected_resources)""" + max_trailing_gap_intervals: int = Field(default=0, ge=0, strict=True) + """Maximum number of trailing scheduler intervals that promotion may accept as debt.""" + planned_frontiers: t.Dict[str, PromotionFrontier] = Field(default_factory=dict) + """The production snapshot frontier captured when the plan was built, keyed by model name.""" @cached_property def start(self) -> TimeLike: @@ -299,6 +304,8 @@ def to_evaluatable(self) -> EvaluatablePlan: environment_statements=self.context_diff.environment_statements, user_provided_flags=self.user_provided_flags, selected_models=self.selected_models, + max_trailing_gap_intervals=self.max_trailing_gap_intervals, + planned_frontiers=dict(self.planned_frontiers), ) @cached_property @@ -338,6 +345,8 @@ class EvaluatablePlan(PydanticModel): environment_statements: t.Optional[t.List[EnvironmentStatements]] = None user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None selected_models: t.Optional[t.Set[str]] = None + max_trailing_gap_intervals: int = Field(default=0, ge=0, strict=True) + planned_frontiers: t.Dict[str, PromotionFrontier] = Field(default_factory=dict) def is_selected_for_backfill(self, model_fqn: str) -> bool: return self.models_to_backfill is None or model_fqn in self.models_to_backfill diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py index f2f432a97e..be113b613a 100644 --- a/sqlmesh/core/plan/evaluator.py +++ b/sqlmesh/core/plan/evaluator.py @@ -363,11 +363,16 @@ def visit_restatement_stage( def visit_environment_record_update_stage( self, stage: stages.EnvironmentRecordUpdateStage, plan: EvaluatablePlan ) -> None: - self.state_sync.promote( - plan.environment, - no_gaps_snapshot_names=stage.no_gaps_snapshot_names if plan.no_gaps else set(), - environment_statements=plan.environment_statements, - ) + promote_kwargs: t.Dict[str, t.Any] = { + "no_gaps_snapshot_names": (stage.no_gaps_snapshot_names if plan.no_gaps else set()), + "environment_statements": plan.environment_statements, + } + if stage.max_trailing_gap_intervals: + promote_kwargs.update( + max_trailing_gap_intervals=stage.max_trailing_gap_intervals, + planned_frontiers=stage.planned_frontiers, + ) + self.state_sync.promote(plan.environment, **promote_kwargs) def visit_migrate_schemas_stage( self, stage: stages.MigrateSchemasStage, plan: EvaluatablePlan diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py index 729e1705b4..b69b9c161d 100644 --- a/sqlmesh/core/plan/stages.py +++ b/sqlmesh/core/plan/stages.py @@ -1,10 +1,11 @@ import typing as t -from dataclasses import dataclass +from dataclasses import dataclass, field from sqlmesh.core import constants as c from sqlmesh.core.environment import EnvironmentStatements, EnvironmentNamingInfo, Environment from sqlmesh.core.plan.common import should_force_rebuild from sqlmesh.core.plan.definition import EvaluatablePlan +from sqlmesh.core.promotion import PromotionFrontier from sqlmesh.core.state_sync import StateReader from sqlmesh.core.scheduler import merged_missing_intervals, SnapshotToIntervals from sqlmesh.core.snapshot.definition import ( @@ -141,9 +142,14 @@ class EnvironmentRecordUpdateStage: Args: no_gaps_snapshot_names: Names of snapshots for which there should be no interval gaps. + max_trailing_gap_intervals: Maximum trailing scheduler intervals that may be accepted + during promotion. + planned_frontiers: Production snapshot frontiers captured when the plan was built. """ no_gaps_snapshot_names: t.Set[str] + max_trailing_gap_intervals: int = 0 + planned_frontiers: t.Dict[str, PromotionFrontier] = field(default_factory=dict) @dataclass @@ -370,7 +376,11 @@ 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}, + max_trailing_gap_intervals=( + plan.max_trailing_gap_intervals if plan.is_prod and plan.no_gaps else 0 + ), + planned_frontiers=(plan.planned_frontiers if plan.is_prod and plan.no_gaps else {}), ) ) diff --git a/sqlmesh/core/promotion.py b/sqlmesh/core/promotion.py new file mode 100644 index 0000000000..3893944e2c --- /dev/null +++ b/sqlmesh/core/promotion.py @@ -0,0 +1,339 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Helpers for validating promotion-time coverage changes.""" + +from __future__ import annotations + +import typing as t +from dataclasses import dataclass +from enum import Enum + +from sqlmesh.core.snapshot import Snapshot, SnapshotId, merge_intervals +from sqlmesh.core.snapshot.definition import Interval +from sqlmesh.utils.date import make_exclusive, now_timestamp, to_timestamp +from sqlmesh.utils.pydantic import PydanticModel + + +class PromotionFrontier(PydanticModel, frozen=True): + """The incumbent snapshot frontier captured while a plan is built. + + ``end`` is an exclusive epoch-millisecond boundary. Keeping the snapshot + identity with the timestamp prevents a plan from applying its old frontier + to a different incumbent snapshot after another plan has run. + """ + + snapshot_id: SnapshotId + end: int + + +class TrailingGap(PydanticModel): + """An accepted contiguous suffix of incumbent coverage absent from a candidate.""" + + start: int + end: int + interval_count: int + next_scheduler_boundary: int + + @property + def interval(self) -> Interval: + return (self.start, self.end) + + +class TrailingGapRejectionReason(str, Enum): + """Stable policy categories for a rejected bounded trailing gap.""" + + DISABLED = "disabled" + FRONTIER_UNAVAILABLE = "frontier_unavailable" + SCHEDULE_INELIGIBLE = "schedule_ineligible" + PLANNED_COVERAGE_INCOMPLETE = "planned_coverage_incomplete" + NON_TRAILING_GAP = "non_trailing_gap" + BOUND_EXCEEDED = "bound_exceeded" + DOWNSTREAM_COVERAGE = "downstream_coverage" + + +@dataclass(frozen=True) +class TrailingGapEvaluation: + """Result of evaluating the bounded trailing-gap promotion policy.""" + + gap: t.Optional[TrailingGap] + reason: t.Optional[TrailingGapRejectionReason] = None + + +def classify_trailing_gap( + target_snapshot: Snapshot, + incumbent_snapshot: Snapshot, + planned_frontier: t.Optional[PromotionFrontier], + max_trailing_gap_intervals: int, + *, + start: int, +) -> t.Optional[TrailingGap]: + """Classify an absent suffix using stored snapshot coverage. + + The existing promotion check deliberately asks ``Snapshot.missing_intervals`` + for the model's required work. That method expands missing work for + lookback, so its result cannot be used as a safety-policy count. This + helper instead subtracts the candidate's *stored* intervals from the + incumbent's stored intervals and then counts the resulting suffix against + the model's scheduler cron boundaries. + + ``None`` means that the change cannot be proven to be a bounded trailing + suffix. Callers should retain the normal strict no-gap failure in that + case. + + Args: + target_snapshot: The candidate snapshot being promoted. + incumbent_snapshot: The snapshot currently referenced by the target + environment. + planned_frontier: The incumbent identity and end captured when the + plan's original backfill bounds were selected. + max_trailing_gap_intervals: Maximum number of scheduler boundaries the + accepted suffix may span. + start: Inclusive start of the original no-gap obligation. The + promotion facade passes this value so historical coverage is + checked independently of lookback. + """ + return evaluate_trailing_gap( + target_snapshot, + incumbent_snapshot, + planned_frontier, + max_trailing_gap_intervals, + start=start, + ).gap + + +def evaluate_trailing_gap( + target_snapshot: Snapshot, + incumbent_snapshot: Snapshot, + planned_frontier: t.Optional[PromotionFrontier], + max_trailing_gap_intervals: int, + *, + start: int, +) -> TrailingGapEvaluation: + """Evaluate an absent suffix and return its bounded-policy reason. + + :func:`classify_trailing_gap` is a convenience wrapper that returns only + the optional accepted gap. Callers that report policy decisions should use + this function so the policy is evaluated exactly once and a rejected gap + is assigned a bounded, stable category. + """ + if max_trailing_gap_intervals < 1: + return _rejected(TrailingGapRejectionReason.DISABLED) + + # The frontier is immutable plan metadata. A matching timestamp on a + # different incumbent snapshot is not sufficient evidence for acceptance. + if planned_frontier is None or planned_frontier.snapshot_id != incumbent_snapshot.snapshot_id: + return _rejected(TrailingGapRejectionReason.FRONTIER_UNAVAILABLE) + + if not _same_schedule(target_snapshot, incumbent_snapshot): + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + + incumbent_intervals = merge_intervals(incumbent_snapshot.intervals) + target_intervals = merge_intervals(target_snapshot.intervals) + if not incumbent_intervals: + return _rejected(TrailingGapRejectionReason.NON_TRAILING_GAP) + + observed_end = incumbent_intervals[-1][1] + planned_end = int(planned_frontier.end) + if planned_end <= 0 or planned_end >= observed_end: + return _rejected(TrailingGapRejectionReason.FRONTIER_UNAVAILABLE) + + required_start = to_timestamp(start) + if required_start >= planned_end: + return _rejected(TrailingGapRejectionReason.FRONTIER_UNAVAILABLE) + + # The candidate must be complete through the original plan obligation. + # This is an actual coverage check and therefore does not include lookback + # expansion. + if not _covers(target_intervals, required_start, planned_end): + return _rejected(TrailingGapRejectionReason.PLANNED_COVERAGE_INCOMPLETE) + + # The incumbent itself must provide a continuous source suffix. A hole in + # incumbent coverage cannot be safely attributed to scheduler drift. + if not _covers(incumbent_intervals, planned_end, observed_end): + return _rejected(TrailingGapRejectionReason.NON_TRAILING_GAP) + + missing = _subtract_coverage( + incumbent_intervals, + target_intervals, + required_start, + observed_end, + ) + if len(missing) != 1: + return _rejected(TrailingGapRejectionReason.NON_TRAILING_GAP) + + gap_start, gap_end = missing[0] + if gap_start < planned_end or gap_end != observed_end: + return _rejected(TrailingGapRejectionReason.NON_TRAILING_GAP) + # A model end is an exclusive execution boundary. A suffix beyond it + # cannot be repaired by the candidate's scheduler, so fail closed even if + # the incumbent happens to contain that coverage. + if target_snapshot.node.end: + try: + model_end = to_timestamp(make_exclusive(target_snapshot.node.end)) + except Exception: + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + if gap_end > model_end: + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + # Coverage after the gap, including coverage beyond the incumbent's + # observed frontier, makes this an internal hole rather than a suffix. + if any(interval_start >= gap_start for interval_start, _ in target_intervals): + return _rejected(TrailingGapRejectionReason.NON_TRAILING_GAP) + + boundary_evaluation = _scheduler_boundaries( + target_snapshot, gap_start, gap_end, max_trailing_gap_intervals + ) + if boundary_evaluation.bound_exceeded: + return _rejected(TrailingGapRejectionReason.BOUND_EXCEEDED) + boundaries = boundary_evaluation.boundaries + if boundaries is None or not boundaries: + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + if len(boundaries) > max_trailing_gap_intervals: + return _rejected(TrailingGapRejectionReason.BOUND_EXCEEDED) + + try: + # The warning describes the next scheduler run from the current wall + # clock. The observed frontier is historical state and may be stale. + next_scheduler_boundary = to_timestamp(target_snapshot.node.cron_next(now_timestamp())) + except Exception: + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + + if next_scheduler_boundary <= observed_end: + return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) + + return TrailingGapEvaluation( + gap=TrailingGap( + start=gap_start, + end=gap_end, + interval_count=len(boundaries), + next_scheduler_boundary=next_scheduler_boundary, + ) + ) + + +def _rejected(reason: TrailingGapRejectionReason) -> TrailingGapEvaluation: + return TrailingGapEvaluation(gap=None, reason=reason) + + +def _same_schedule(target_snapshot: Snapshot, incumbent_snapshot: Snapshot) -> bool: + """Require the two snapshots to retain the same interval semantics.""" + return ( + target_snapshot.node.cron == incumbent_snapshot.node.cron + and target_snapshot.node.cron_tz == incumbent_snapshot.node.cron_tz + and target_snapshot.node.interval_unit == incumbent_snapshot.node.interval_unit + and target_snapshot.allow_partials == incumbent_snapshot.allow_partials + and target_snapshot.node.start == incumbent_snapshot.node.start + and target_snapshot.node.end == incumbent_snapshot.node.end + ) + + +def _covers(intervals: t.Sequence[Interval], start: int, end: int) -> bool: + if start >= end: + return True + + cursor = start + for interval_start, interval_end in intervals: + if interval_end <= cursor: + continue + if interval_start > cursor: + return False + cursor = max(cursor, interval_end) + if cursor >= end: + return True + return False + + +def _subtract_coverage( + source: t.Sequence[Interval], + covered: t.Sequence[Interval], + start: int, + end: int, +) -> t.List[Interval]: + """Return uncovered portions of source in ``[start, end)``.""" + missing: t.List[Interval] = [] + covered_index = 0 + + for source_start, source_end in source: + source_start = max(source_start, start) + source_end = min(source_end, end) + if source_start >= source_end: + continue + + cursor = source_start + while covered_index < len(covered) and covered[covered_index][1] <= cursor: + covered_index += 1 + + index = covered_index + while index < len(covered): + covered_start, covered_end = covered[index] + if covered_start >= source_end: + break + if covered_end <= cursor: + index += 1 + continue + if covered_start > cursor: + missing.append((cursor, min(covered_start, source_end))) + cursor = max(cursor, covered_end) + if cursor >= source_end: + break + index += 1 + + if cursor < source_end: + missing.append((cursor, source_end)) + + return missing + + +@dataclass(frozen=True) +class _SchedulerBoundaryEvaluation: + boundaries: t.Optional[t.List[Interval]] + bound_exceeded: bool = False + + +def _scheduler_boundaries( + snapshot: Snapshot, + start: int, + end: int, + max_count: int, +) -> _SchedulerBoundaryEvaluation: + """Evaluate scheduler alignment while preserving bound failures separately.""" + if start >= end: + return _SchedulerBoundaryEvaluation(boundaries=[]) + + node = snapshot.node + interval_unit = node.interval_unit + + def coverage_boundary(cron_timestamp: int) -> int: + # Match Snapshot.missing_intervals: non-partial coverage ends at the + # interval-unit floor of the scheduler tick, including offset crons. + return ( + cron_timestamp + if snapshot.allow_partials + else to_timestamp(interval_unit.cron_floor(cron_timestamp)) + ) + + try: + cron_timestamp = to_timestamp(node.cron_floor(start)) + if coverage_boundary(cron_timestamp) < start: + cron_timestamp = to_timestamp(node.cron_next(start)) + if coverage_boundary(cron_timestamp) != start: + return _SchedulerBoundaryEvaluation(boundaries=None) + except Exception: + return _SchedulerBoundaryEvaluation(boundaries=None) + + boundaries: t.List[Interval] = [] + current = start + while current < end: + if len(boundaries) >= max_count: + return _SchedulerBoundaryEvaluation(boundaries=None, bound_exceeded=True) + try: + next_cron_timestamp = to_timestamp(node.cron_next(cron_timestamp)) + next_boundary = coverage_boundary(next_cron_timestamp) + except Exception: + return _SchedulerBoundaryEvaluation(boundaries=None) + if next_boundary <= current or next_boundary > end: + return _SchedulerBoundaryEvaluation(boundaries=None) + boundaries.append((current, next_boundary)) + current = next_boundary + cron_timestamp = next_cron_timestamp + + return _SchedulerBoundaryEvaluation(boundaries=boundaries if current == end else None) diff --git a/sqlmesh/core/state_sync/base.py b/sqlmesh/core/state_sync/base.py index 5c35be5ccb..6fdd88a37c 100644 --- a/sqlmesh/core/state_sync/base.py +++ b/sqlmesh/core/state_sync/base.py @@ -34,6 +34,7 @@ PromotionResult, ExpiredBatchRange, ) +from sqlmesh.core.promotion import PromotionFrontier logger = logging.getLogger(__name__) @@ -416,6 +417,8 @@ def promote( environment: Environment, no_gaps_snapshot_names: t.Optional[t.Set[str]] = None, environment_statements: t.Optional[t.List[EnvironmentStatements]] = None, + max_trailing_gap_intervals: int = 0, + planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None, ) -> PromotionResult: """Update the environment to reflect the current state. @@ -427,6 +430,12 @@ def promote( all snapshots will be checked. The data gap check ensures that models that are already a part of the target environment have no data gaps when compared against previous snapshots for same models. + max_trailing_gap_intervals: Maximum number of scheduler boundaries + of trailing incumbent coverage that may be accepted. Zero + preserves strict no-gap behavior. + planned_frontiers: The incumbent snapshot identities and coverage + frontiers captured while the plan was built, keyed by model + name. Returns: A tuple of (added snapshot table infos, removed snapshot table infos) diff --git a/sqlmesh/core/state_sync/db/facade.py b/sqlmesh/core/state_sync/db/facade.py index 572e54b7f1..b3f393b2d0 100644 --- a/sqlmesh/core/state_sync/db/facade.py +++ b/sqlmesh/core/state_sync/db/facade.py @@ -24,6 +24,7 @@ from sqlmesh.core.console import Console, get_console +from sqlmesh.core import constants as c from sqlmesh.core.engine_adapter import EngineAdapter from sqlmesh.core.environment import Environment, EnvironmentStatements, EnvironmentSummary from sqlmesh.core.snapshot import ( @@ -62,6 +63,12 @@ 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.core.promotion import ( + PromotionFrontier, + TrailingGap, + TrailingGapRejectionReason, + evaluate_trailing_gap, +) from sqlmesh.utils.date import TimeLike, to_timestamp, time_like_to_str, now_timestamp from sqlmesh.utils.errors import ConflictingPlanError, SQLMeshError @@ -149,6 +156,8 @@ def promote( environment: Environment, no_gaps_snapshot_names: t.Optional[t.Set[str]] = None, environment_statements: t.Optional[t.List[EnvironmentStatements]] = None, + max_trailing_gap_intervals: int = 0, + planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None, ) -> PromotionResult: """Update the environment to reflect the current state. @@ -160,6 +169,12 @@ def promote( all snapshots will be checked. The data gap check ensures that models that are already a part of the target environment have no data gaps when compared against previous snapshots for same models. + max_trailing_gap_intervals: Maximum number of scheduler boundaries + of trailing incumbent coverage that may be accepted. Zero + preserves strict no-gap behavior. + planned_frontiers: The incumbent snapshot identities and coverage + frontiers captured while the plan was built, keyed by model + name. Returns: A tuple of (added snapshot table infos, removed snapshot table infos, and environment target suffix for the removed table infos) @@ -174,6 +189,9 @@ def promote( f"Missing snapshots {missing}. Make sure to push and backfill your snapshots." ) + accepted_trailing_gaps: t.List[ + t.Tuple[Snapshot, Snapshot, PromotionFrontier, TrailingGap] + ] = [] existing_environment = self.environment_state.get_environment( environment.name, lock_for_update=True ) @@ -203,10 +221,15 @@ def promote( ) if no_gaps_snapshot_names != set(): snapshots = self.get_snapshots(environment.snapshots).values() - self._ensure_no_gaps( + accepted_trailing_gaps = self._ensure_no_gaps( snapshots, existing_environment, no_gaps_snapshot_names, + max_trailing_gap_intervals=max_trailing_gap_intervals + if environment.name == c.PROD + else 0, + planned_frontiers=planned_frontiers, + candidate_plan_id=environment.plan_id, ) demoted_snapshots = set(existing_environment.snapshots) - set(environment.snapshots) # Update the updated_at attribute. @@ -234,7 +257,7 @@ def promote( views_that_changed_location ) - return PromotionResult( + result = PromotionResult( added=sorted(added_table_infos), removed=list(removed), removed_environment_naming_info=( @@ -242,6 +265,31 @@ def promote( ), ) + for target_snapshot, incumbent_snapshot, frontier, gap in accepted_trailing_gaps: + payload = { + "environment": environment.name, + "plan_id": environment.plan_id, + "previous_plan_id": environment.previous_plan_id, + "model": target_snapshot.name, + "incumbent_snapshot_id": str(incumbent_snapshot.snapshot_id), + "candidate_snapshot_id": str(target_snapshot.snapshot_id), + "planned_frontier": frontier.end, + "observed_frontier": gap.end, + "gap_count": gap.interval_count, + "gap_range": gap.interval, + "next_scheduler_boundary": gap.next_scheduler_boundary, + } + logger.warning( + "Accepted bounded trailing gap during promotion: %s", + payload, + extra={ + "sqlmesh_event": "bounded_trailing_gap_accepted", + "sqlmesh_promotion": payload, + }, + ) + + return result + @transactional() def finalize(self, environment: Environment) -> None: """Finalize the target environment, indicating that this environment has been @@ -585,7 +633,11 @@ def _ensure_no_gaps( target_snapshots: t.Iterable[Snapshot], target_environment: Environment, snapshot_names: t.Optional[t.Set[str]], - ) -> None: + *, + max_trailing_gap_intervals: int = 0, + planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None, + candidate_plan_id: t.Optional[str] = None, + ) -> t.List[t.Tuple[Snapshot, Snapshot, PromotionFrontier, TrailingGap]]: target_snapshots_by_name = {s.name: s for s in target_snapshots} changed_version_prev_snapshots_by_name = { @@ -599,6 +651,9 @@ def _ensure_no_gaps( changed_version_prev_snapshots_by_name.values() ).values() cache: t.Dict[str, datetime] = {} + accepted_trailing_gaps: t.List[ + t.Tuple[Snapshot, Snapshot, PromotionFrontier, TrailingGap] + ] = [] for prev_snapshot in prev_snapshots: target_snapshot = target_snapshots_by_name[prev_snapshot.name] @@ -619,11 +674,51 @@ def _ensure_no_gaps( ) if missing_intervals: + frontier = (planned_frontiers or {}).get(prev_snapshot.name) + evaluation = evaluate_trailing_gap( + target_snapshot, + prev_snapshot, + frontier, + max_trailing_gap_intervals, + start=start, + ) + if evaluation.gap and frontier: + accepted_trailing_gaps.append( + (target_snapshot, prev_snapshot, frontier, evaluation.gap) + ) + continue + + _log_trailing_gap_rejection( + target_environment, + target_snapshot, + evaluation.reason, + candidate_plan_id=candidate_plan_id, + ) + raise SQLMeshError( f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " "Please re-apply your plan to resolve this error." ) + for target_snapshot, _, _, trailing_gap in accepted_trailing_gaps: + if _has_downstream_coverage( + target_snapshots_by_name, + target_snapshot.name, + trailing_gap, + ): + _log_trailing_gap_rejection( + target_environment, + target_snapshot, + TrailingGapRejectionReason.DOWNSTREAM_COVERAGE, + candidate_plan_id=candidate_plan_id, + ) + raise SQLMeshError( + f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " + "Please re-apply your plan to resolve this error." + ) + + return accepted_trailing_gaps + @contextlib.contextmanager def _transaction(self) -> t.Iterator[None]: with self.engine_adapter.transaction(): @@ -652,3 +747,66 @@ def _remove_partial_intervals( snapshot_id, ) return results + + +def _log_trailing_gap_rejection( + target_environment: Environment, + target_snapshot: Snapshot, + reason: t.Optional[TrailingGapRejectionReason], + *, + candidate_plan_id: t.Optional[str], +) -> None: + """Report a rejected policy decision without changing deployment behavior.""" + try: + payload = { + "environment": target_environment.name, + "plan_id": candidate_plan_id, + "model": target_snapshot.name, + "candidate_snapshot_id": str(target_snapshot.snapshot_id), + "reason": reason.value if reason else None, + } + logger.warning( + "Rejected bounded trailing gap during promotion: %s", + payload, + extra={ + "sqlmesh_event": "bounded_trailing_gap_rejected", + "sqlmesh_promotion": payload, + }, + ) + except Exception: + # Observability must not mask the policy's existing SQLMeshError. + pass + + +def _has_downstream_coverage( + snapshots_by_name: t.Mapping[str, Snapshot], model_name: str, gap: TrailingGap +) -> bool: + """Return whether a target descendant already covers an accepted parent tail. + + A descendant interval built before its parent's accepted tail is repaired + can be marked complete against incomplete input. Keeping this check + conservative avoids accepting that state when auto-upstream selection or + a later scheduler run cannot invalidate the descendant interval. + """ + children_by_parent: t.Dict[str, t.Set[str]] = {} + for snapshot in snapshots_by_name.values(): + for parent in snapshot.parents: + children_by_parent.setdefault(parent.name, set()).add(snapshot.name) + + descendants: t.Set[str] = set() + pending = [model_name] + while pending: + parent_name = pending.pop() + for child_name in children_by_parent.get(parent_name, set()): + if child_name not in descendants: + descendants.add(child_name) + pending.append(child_name) + + for descendant_name in descendants: + descendant = snapshots_by_name[descendant_name] + if any( + interval_start < gap.end and interval_end > gap.start + for interval_start, interval_end in descendant.intervals + ): + return True + return False diff --git a/sqlmesh/integrations/github/cicd/config.py b/sqlmesh/integrations/github/cicd/config.py index 2a2e2efa49..b5570ed7b5 100644 --- a/sqlmesh/integrations/github/cicd/config.py +++ b/sqlmesh/integrations/github/cicd/config.py @@ -31,6 +31,7 @@ class GithubCICDBotConfig(BaseConfig): skip_pr_backfill_: t.Optional[bool] = Field(default=None, alias="skip_pr_backfill") pr_include_unmodified_: t.Optional[bool] = Field(default=None, alias="pr_include_unmodified") run_on_deploy_to_prod: bool = False + max_trailing_gap_intervals_on_deploy_to_prod: int = Field(default=0, ge=0, strict=True) pr_environment_name: t.Optional[str] = None pr_min_intervals: t.Optional[int] = None pr_preview_min_intervals: int = Field(default=1, ge=0) diff --git a/sqlmesh/integrations/github/cicd/controller.py b/sqlmesh/integrations/github/cicd/controller.py index 58b0664c2d..07e077f38e 100644 --- a/sqlmesh/integrations/github/cicd/controller.py +++ b/sqlmesh/integrations/github/cicd/controller.py @@ -439,6 +439,7 @@ def prod_plan(self) -> Plan: categorizer_config=self.bot_config.auto_categorize_changes, run=self.bot_config.run_on_deploy_to_prod, forward_only=self.forward_only_plan, + max_trailing_gap_intervals=self.bot_config.max_trailing_gap_intervals_on_deploy_to_prod, ) assert self._prod_plan_builder return self._prod_plan_builder.build() diff --git a/tests/core/state_sync/test_state_sync.py b/tests/core/state_sync/test_state_sync.py index 503f89296e..6d684b79a3 100644 --- a/tests/core/state_sync/test_state_sync.py +++ b/tests/core/state_sync/test_state_sync.py @@ -24,6 +24,7 @@ SeedModel, SqlModel, ) +from sqlmesh.core.promotion import PromotionFrontier from sqlmesh.core.snapshot import ( Snapshot, SnapshotChangeCategory, @@ -49,7 +50,7 @@ RowBoundary, ) from sqlmesh.utils.date import now_timestamp, to_datetime, to_timestamp -from sqlmesh.utils.errors import SQLMeshError, StateMigrationError +from sqlmesh.utils.errors import ConflictingPlanError, SQLMeshError, StateMigrationError pytestmark = pytest.mark.slow @@ -108,6 +109,10 @@ def promote_snapshots( environment: str, no_gaps: bool = False, no_gaps_snapshot_names: t.Optional[t.Set[str]] = None, + max_trailing_gap_intervals: int = 0, + planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None, + plan_id: str = "test_plan_id", + previous_plan_id: t.Optional[str] = "test_plan_id", environment_suffix_target: EnvironmentSuffixTarget = EnvironmentSuffixTarget.SCHEMA, environment_catalog_mapping: t.Optional[t.Dict[re.Pattern, str]] = None, ) -> PromotionResult: @@ -118,12 +123,166 @@ def promote_snapshots( snapshots=[snapshot.table_info for snapshot in snapshots], start_at="2022-01-01", end_at="2022-01-01", - plan_id="test_plan_id", - previous_plan_id="test_plan_id", + plan_id=plan_id, + previous_plan_id=previous_plan_id, ) return state_sync.promote( - env, no_gaps_snapshot_names=no_gaps_snapshot_names if no_gaps else set() + env, + no_gaps_snapshot_names=no_gaps_snapshot_names if no_gaps else set(), + max_trailing_gap_intervals=max_trailing_gap_intervals, + planned_frontiers=planned_frontiers, + ) + + +def _prepare_trailing_gap_promotion( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +) -> t.Tuple[Snapshot, Snapshot, PromotionFrontier]: + model = SqlModel( + name="a", + query=parse_one("select 1, ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@hourly", + start="2023-01-01", + ) + incumbent = make_snapshot(model, version="a") + incumbent.change_category = SnapshotChangeCategory.BREAKING + state_sync.push_snapshots([incumbent]) + state_sync.add_interval(incumbent, "2023-01-01 00:00", "2023-01-01 11:00") + promote_snapshots(state_sync, [incumbent], "prod") + + target = make_snapshot(model.copy(), version="b") + target.change_category = SnapshotChangeCategory.BREAKING + target.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "target"}) + state_sync.push_snapshots([target]) + state_sync.add_interval(target, "2023-01-01 00:00", "2023-01-01 11:00") + + # Simulate an overlapping scheduler run after the production plan captured + # its original frontier at 11:00. + state_sync.add_interval(incumbent, "2023-01-01 11:00", "2023-01-01 12:00") + frontier = PromotionFrontier( + snapshot_id=incumbent.snapshot_id, + end=to_timestamp("2023-01-01 11:00"), + ) + return incumbent, target, frontier + + +def _prepare_multi_model_trailing_gap_promotion( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +) -> t.Tuple[t.List[Snapshot], t.List[Snapshot], t.Dict[str, PromotionFrontier]]: + models = [ + SqlModel( + name=name, + query=parse_one("select 1, ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@hourly", + start="2023-01-01", + ) + for name in ("a", "b") + ] + incumbents = [make_snapshot(model, version="a") for model in models] + for snapshot in incumbents: + snapshot.change_category = SnapshotChangeCategory.BREAKING + state_sync.push_snapshots(incumbents) + for snapshot in incumbents: + state_sync.add_interval(snapshot, "2023-01-01 00:00", "2023-01-01 11:00") + promote_snapshots(state_sync, incumbents, "prod") + + targets = [] + for incumbent, model in zip(incumbents, models): + target = make_snapshot(model.copy(), version="b") + target.change_category = SnapshotChangeCategory.BREAKING + target.fingerprint = incumbent.fingerprint.copy( + update={"data_hash": f"{target.name}_target"} + ) + targets.append(target) + state_sync.push_snapshots(targets) + state_sync.add_interval(targets[0], "2023-01-01 00:00", "2023-01-01 11:00") + # This model has a valid one-interval suffix. + state_sync.add_interval(incumbents[0], "2023-01-01 11:00", "2023-01-01 12:00") + # This model has a two-interval suffix, which exceeds the bound of one. + state_sync.add_interval(targets[1], "2023-01-01 00:00", "2023-01-01 10:00") + state_sync.add_interval(incumbents[1], "2023-01-01 11:00", "2023-01-01 12:00") + + frontiers = { + incumbents[0].name: PromotionFrontier( + snapshot_id=incumbents[0].snapshot_id, + end=to_timestamp("2023-01-01 11:00"), + ), + incumbents[1].name: PromotionFrontier( + snapshot_id=incumbents[1].snapshot_id, + end=to_timestamp("2023-01-01 10:00"), + ), + } + return incumbents, targets, frontiers + + +def _prepare_parent_gap_with_descendants( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + *, + descendant_coverage: bool, + transitive: bool = False, +) -> t.Tuple[t.List[Snapshot], Snapshot, PromotionFrontier]: + parent_model = SqlModel( + name="parent", + query=parse_one("select 1, ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@hourly", + start="2023-01-01", + ) + child_model = SqlModel( + name="child", + query=parse_one("select * from parent"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@hourly", + start="2023-01-01", + ) + grandchild_model = SqlModel( + name="grandchild", + query=parse_one("select * from child"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@hourly", + start="2023-01-01", ) + nodes = { + parent_model.fqn: parent_model, + child_model.fqn: child_model, + grandchild_model.fqn: grandchild_model, + } + + incumbent = make_snapshot(parent_model, version="a", nodes=nodes) + child = make_snapshot(child_model, version="a", nodes=nodes) + snapshots = [incumbent, child] + if transitive: + snapshots.append(make_snapshot(grandchild_model, version="a", nodes=nodes)) + for snapshot in snapshots: + snapshot.change_category = SnapshotChangeCategory.BREAKING + state_sync.push_snapshots(snapshots) + + for snapshot in snapshots: + has_stale_descendant_coverage = descendant_coverage and ( + not transitive or snapshot is snapshots[-1] + ) + state_sync.add_interval( + snapshot, + "2023-01-01 00:00", + "2023-01-01 12:00" if has_stale_descendant_coverage else "2023-01-01 11:00", + ) + promote_snapshots(state_sync, snapshots, "prod") + + target = make_snapshot(parent_model.copy(), version="b", nodes=nodes) + target.change_category = SnapshotChangeCategory.BREAKING + target.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "target"}) + state_sync.push_snapshots([target]) + state_sync.add_interval(target, "2023-01-01 00:00", "2023-01-01 11:00") + state_sync.add_interval(incumbent, "2023-01-01 11:00", "2023-01-01 12:00") + + frontier = PromotionFrontier( + snapshot_id=incumbent.snapshot_id, + end=to_timestamp("2023-01-01 11:00"), + ) + target_snapshots = [target, *snapshots[1:]] + return target_snapshots, target, frontier def delete_versions(state_sync: EngineAdapterStateSync) -> None: @@ -1052,6 +1211,304 @@ def test_promote_snapshots_no_gaps_lookback( promote_snapshots(state_sync, [new_snapshot_same_version], "prod", no_gaps=True) +def test_promote_snapshots_accepts_bounded_trailing_gap( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + caplog: pytest.LogCaptureFixture, +): + incumbent, target, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + + with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + environment = state_sync.get_environment("prod") + assert environment is not None + assert environment.snapshots[0].snapshot_id == target.snapshot_id + assert state_sync.get_snapshots([target])[target.snapshot_id].intervals == [ + (to_timestamp("2023-01-01 00:00"), to_timestamp("2023-01-01 11:00")) + ] + + accepted_warnings = [ + record + for record in caplog.records + if getattr(record, "sqlmesh_event", None) == "bounded_trailing_gap_accepted" + ] + assert len(accepted_warnings) == 1 + payload = getattr(accepted_warnings[0], "sqlmesh_promotion") + assert payload["environment"] == "prod" + assert payload["plan_id"] == "new_plan_id" + assert payload["incumbent_snapshot_id"] == str(incumbent.snapshot_id) + assert payload["candidate_snapshot_id"] == str(target.snapshot_id) + assert payload["planned_frontier"] == to_timestamp("2023-01-01 11:00") + assert payload["observed_frontier"] == to_timestamp("2023-01-01 12:00") + assert payload["gap_count"] == 1 + + +def test_promote_snapshots_rolls_back_multiple_models_before_warning( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + caplog: pytest.LogCaptureFixture, +): + incumbents, targets, frontiers = _prepare_multi_model_trailing_gap_promotion( + state_sync, make_snapshot + ) + caplog.clear() + + with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + targets, + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers=frontiers, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + assert not any( + getattr(record, "sqlmesh_event", None) == "bounded_trailing_gap_accepted" + for record in caplog.records + ) + environment = state_sync.get_environment("prod") + assert environment is not None + assert environment.plan_id == "test_plan_id" + assert {snapshot.snapshot_id for snapshot in environment.snapshots} == { + snapshot.snapshot_id for snapshot in incumbents + } + + +def test_promote_snapshots_keeps_dev_strict_with_positive_bound( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + incumbent, target, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + promote_snapshots(state_sync, [incumbent], "dev") + + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + [target], + "dev", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + +@pytest.mark.parametrize("frontiers", [None, {}]) +def test_promote_snapshots_requires_planned_frontier_for_relaxed_policy( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + frontiers: t.Optional[t.Dict[str, PromotionFrontier]], +): + _, target, _ = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers=frontiers, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + +def test_promote_snapshots_reports_rejection_with_candidate_identity( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + caplog: pytest.LogCaptureFixture, +): + _, target, _ = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + + with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + plan_id="candidate_plan_id", + previous_plan_id="test_plan_id", + ) + + rejected_warnings = [ + record + for record in caplog.records + if getattr(record, "sqlmesh_event", None) == "bounded_trailing_gap_rejected" + ] + assert len(rejected_warnings) == 1 + payload = getattr(rejected_warnings[0], "sqlmesh_promotion") + assert payload == { + "environment": "prod", + "plan_id": "candidate_plan_id", + "model": target.name, + "candidate_snapshot_id": str(target.snapshot_id), + "reason": "disabled", + } + + +def test_promote_snapshots_keeps_policy_error_when_rejection_logging_fails( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + _, target, _ = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + + with patch( + "sqlmesh.core.state_sync.db.facade.logger.warning", + side_effect=RuntimeError("capture failed"), + ): + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + plan_id="candidate_plan_id", + previous_plan_id="test_plan_id", + ) + + +def test_promote_snapshots_reports_downstream_policy_rejection( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + caplog: pytest.LogCaptureFixture, +): + target_snapshots, target, frontier = _prepare_parent_gap_with_descendants( + state_sync, + make_snapshot, + descendant_coverage=True, + ) + + with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + target_snapshots, + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="candidate_plan_id", + previous_plan_id="test_plan_id", + ) + + rejected_warnings = [ + record + for record in caplog.records + if getattr(record, "sqlmesh_event", None) == "bounded_trailing_gap_rejected" + ] + assert len(rejected_warnings) == 1 + payload = getattr(rejected_warnings[0], "sqlmesh_promotion") + assert payload["reason"] == "downstream_coverage" + assert payload["plan_id"] == "candidate_plan_id" + + +def test_promote_snapshots_rejects_frontier_identity_mismatch( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + incumbent, target, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + mismatched_frontier = PromotionFrontier( + snapshot_id=SnapshotId(name=incumbent.name, identifier="different"), + end=frontier.end, + ) + + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: mismatched_frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + +def test_promote_snapshots_conflict_precedes_relaxed_gap_check( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + _, target, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + + with pytest.raises(ConflictingPlanError, match="Another plan"): + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="wrong_plan_id", + ) + + +@pytest.mark.parametrize("transitive", [False, True]) +def test_promote_snapshots_rejects_descendant_coverage_for_accepted_parent_gap( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + transitive: bool, +): + target_snapshots, _, frontier = _prepare_parent_gap_with_descendants( + state_sync, + make_snapshot, + descendant_coverage=True, + transitive=transitive, + ) + + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + promote_snapshots( + state_sync, + target_snapshots, + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target_snapshots[0].name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + +def test_promote_snapshots_accepts_parent_gap_when_descendant_is_missing( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + target_snapshots, target, frontier = _prepare_parent_gap_with_descendants( + state_sync, + make_snapshot, + descendant_coverage=False, + ) + + result = promote_snapshots( + state_sync, + target_snapshots, + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + assert result.removed == [] + environment = state_sync.get_environment("prod") + assert environment is not None + assert target.snapshot_id in {snapshot.snapshot_id for snapshot in environment.snapshots} + + def test_finalize(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable): snapshot_a = make_snapshot( SqlModel( diff --git a/tests/core/test_plan_policy.py b/tests/core/test_plan_policy.py new file mode 100644 index 0000000000..11c1a7d991 --- /dev/null +++ b/tests/core/test_plan_policy.py @@ -0,0 +1,384 @@ +# SPDX-License-Identifier: Apache-2.0 + +from datetime import datetime +from pathlib import Path + +import pytest +import time_machine +from sqlglot import parse_one +from pydantic import ValidationError + +from sqlmesh.core.context import Context +from sqlmesh.core.context_diff import ContextDiff +from sqlmesh.core.environment import Environment +from sqlmesh.core.model import IncrementalByTimeRangeKind, SqlModel +from sqlmesh.core.plan import BuiltInPlanEvaluator, PlanBuilder +from sqlmesh.core.plan.definition import EvaluatablePlan +from sqlmesh.core.plan.stages import EnvironmentRecordUpdateStage, build_plan_stages +from sqlmesh.core.promotion import PromotionFrontier +from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory +from sqlmesh.utils.date import to_timestamp + + +def _context_diff(current: Snapshot, previous: Snapshot) -> ContextDiff: + return ContextDiff( + environment="prod", + is_new_environment=False, + is_unfinalized_environment=False, + normalize_environment_name=True, + create_from="prod", + create_from_env_exists=True, + added=set(), + removed_snapshots={}, + modified_snapshots={current.name: (current, previous)}, + snapshots={current.snapshot_id: current}, + new_snapshots={current.snapshot_id: current}, + previous_plan_id="previous-plan", + previously_promoted_snapshot_ids={previous.snapshot_id}, + previous_finalized_snapshots=None, + previous_gateway_managed_virtual_layer=False, + gateway_managed_virtual_layer=False, + environment_statements=[], + ) + + +def _write_hourly_model(path: Path, value: int) -> None: + (path / "models").mkdir(exist_ok=True) + (path / "config.yaml").write_text("model_defaults:\n dialect: duckdb\n") + (path / "models" / "model.sql").write_text( + f""" +MODEL ( + name test.model, + kind INCREMENTAL_BY_TIME_RANGE (time_column ts, batch_size 1), + cron '@hourly', + start '2024-01-01' +); +SELECT ts, {value} AS value +FROM generate_series(@start_dt, @end_dt, INTERVAL '1 hour') AS t(ts) +""" + ) + + +def _build_plan_with_choices(builder: PlanBuilder): + plan = builder.build() + for snapshot in plan.uncategorized: + builder.set_choice(snapshot, SnapshotChangeCategory.BREAKING) + return builder.build() + + +def test_planned_frontier_uses_modified_incumbent_identity(make_snapshot): + previous = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + ), + version="previous", + ) + current = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds, 1 as value"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + ), + version="current", + ) + previous.categorize_as(SnapshotChangeCategory.BREAKING) + current.categorize_as(SnapshotChangeCategory.BREAKING) + context_diff = _context_diff(current, previous) + + frontier_end = datetime(2024, 1, 2) + frontiers = Context._get_planned_frontiers(context_diff, {current.name: frontier_end}) + + assert frontiers == { + current.name: PromotionFrontier( + snapshot_id=previous.snapshot_id, + end=to_timestamp(frontier_end), + ) + } + + no_previous_plan_diff = context_diff.copy(update={"previous_plan_id": None}) + assert Context._get_planned_frontiers(no_previous_plan_diff, {current.name: frontier_end}) == {} + + +@pytest.mark.slow +def test_context_frontier_is_captured_before_end_and_execution_overrides(tmp_path): + _write_hourly_model(tmp_path, 1) + + with time_machine.travel("2024-01-01 04:00:00 UTC", tick=False): + context = Context(paths=tmp_path) + context.plan( + "prod", + no_prompts=True, + auto_apply=True, + execution_time="2024-01-01 04:00:00", + ) + + _write_hourly_model(tmp_path, 2) + context.load() + execution_time_plan = _build_plan_with_choices( + context.plan_builder( + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + execution_time="2024-01-01 06:00:00", + ) + ) + assert len(execution_time_plan.planned_frontiers) == 1 + assert next(iter(execution_time_plan.planned_frontiers.values())).end == to_timestamp( + "2024-01-01 04:00:00" + ) + assert to_timestamp(execution_time_plan.end) == to_timestamp("2024-01-01 06:00:00") + assert execution_time_plan.end_override_per_model + assert all( + to_timestamp(end) == to_timestamp("2024-01-01 06:00:00") + for end in execution_time_plan.end_override_per_model.values() + ) + + plan = _build_plan_with_choices( + context.plan_builder( + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + execution_time="2024-01-01 06:00:00", + restate_models=["test.model"], + end="2024-01-01 05:00:00", + always_include_local_changes=True, + ) + ) + + assert plan.max_trailing_gap_intervals == 1 + assert len(plan.planned_frontiers) == 1 + frontier = next(iter(plan.planned_frontiers.values())) + assert frontier.end == to_timestamp("2024-01-01 04:00:00") + assert frontier.end < to_timestamp(plan.end) + # The explicit end is handled by PlanBuilder and intentionally removes per-model + # backfill caps. The separately captured frontier remains available for promotion. + assert plan.end_override_per_model is None + context.close() + + +@pytest.mark.slow +def test_context_run_frontier_capture_does_not_add_backfill_caps(tmp_path): + _write_hourly_model(tmp_path, 1) + + with time_machine.travel("2024-01-01 04:00:00 UTC", tick=False): + context = Context(paths=tmp_path) + context.plan( + "prod", + no_prompts=True, + auto_apply=True, + execution_time="2024-01-01 04:00:00", + ) + + _write_hourly_model(tmp_path, 2) + context.load() + run_plan = _build_plan_with_choices( + context.plan_builder( + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + run=True, + execution_time="2024-01-01 06:00:00", + ) + ) + strict_run_plan = _build_plan_with_choices( + context.plan_builder( + "prod", + no_gaps=True, + run=True, + execution_time="2024-01-01 06:00:00", + ) + ) + + assert len(run_plan.planned_frontiers) == 1 + assert next(iter(run_plan.planned_frontiers.values())).end == to_timestamp( + "2024-01-01 04:00:00" + ) + assert run_plan.end_override_per_model is None + assert strict_run_plan.end_override_per_model is None + assert run_plan.missing_intervals == strict_run_plan.missing_intervals + context.close() + + +def test_evaluatable_plan_round_trip_preserves_policy_and_backfill_overrides(make_snapshot): + snapshot = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + ), + version="current", + ) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + planned_frontier = PromotionFrontier(snapshot_id=snapshot.snapshot_id, end=123) + start_override = datetime(2024, 1, 1) + end_override = datetime(2024, 1, 2) + plan = EvaluatablePlan( + start="2024-01-01", + end="2024-01-03", + new_snapshots=[snapshot], + environment=Environment( + snapshots=[snapshot.table_info], + start_at="2024-01-01", + end_at="2024-01-03", + plan_id="plan", + previous_plan_id="previous-plan", + ), + no_gaps=True, + skip_backfill=False, + empty_backfill=False, + restatements={}, + restate_all_snapshots=False, + is_dev=False, + allow_destructive_models=set(), + allow_additive_models=set(), + forward_only=False, + end_bounded=True, + ensure_finalized_snapshots=False, + directly_modified_snapshots=[snapshot.snapshot_id], + indirectly_modified_snapshots={}, + metadata_updated_snapshots=[], + removed_snapshots=[], + requires_backfill=True, + start_override_per_model={snapshot.name: start_override}, + end_override_per_model={snapshot.name: end_override}, + execution_time="2024-01-03", + disabled_restatement_models=set(), + max_trailing_gap_intervals=1, + planned_frontiers={snapshot.name: planned_frontier}, + ) + + restored = EvaluatablePlan.parse_raw(plan.json()) + + assert restored.max_trailing_gap_intervals == 1 + assert restored.planned_frontiers == {snapshot.name: planned_frontier} + assert restored.start_override_per_model == {snapshot.name: start_override} + assert restored.end_override_per_model == {snapshot.name: end_override} + + +def test_plan_builder_threads_policy_and_frontier(make_snapshot): + previous = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + start="2024-01-01", + ), + version="previous", + ) + current = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds, 1 as value"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + start="2024-01-01", + ), + version="current", + ) + previous.categorize_as(SnapshotChangeCategory.BREAKING) + current.categorize_as(SnapshotChangeCategory.BREAKING) + planned_frontier = PromotionFrontier(snapshot_id=previous.snapshot_id, end=123) + + plan = PlanBuilder( + _context_diff(current, previous), + no_gaps=True, + execution_time="2024-01-03", + max_trailing_gap_intervals=1, + planned_frontiers={current.name: planned_frontier}, + ).build() + + assert plan.max_trailing_gap_intervals == 1 + assert plan.planned_frontiers == {current.name: planned_frontier} + assert plan.to_evaluatable().planned_frontiers == plan.planned_frontiers + + +def test_plan_evaluator_passes_policy_to_promotion(make_snapshot, mocker): + previous = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + start="2024-01-01", + ), + version="previous", + ) + current = make_snapshot( + SqlModel( + name="a", + query=parse_one("select ds, 1 as value"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + start="2024-01-01", + ), + version="current", + ) + previous.categorize_as(SnapshotChangeCategory.BREAKING) + current.categorize_as(SnapshotChangeCategory.BREAKING) + planned_frontier = PromotionFrontier(snapshot_id=previous.snapshot_id, end=123) + plan = PlanBuilder( + _context_diff(current, previous), + no_gaps=True, + execution_time="2024-01-03", + max_trailing_gap_intervals=1, + planned_frontiers={current.name: planned_frontier}, + ).build() + + state_reader = mocker.Mock() + state_reader.get_snapshots.return_value = {} + state_reader.get_environment.return_value = None + evaluatable_plan = plan.to_evaluatable() + stages = build_plan_stages(evaluatable_plan, state_reader, None) + environment_stage = next( + stage for stage in stages if isinstance(stage, EnvironmentRecordUpdateStage) + ) + state_sync = mocker.Mock() + evaluator = BuiltInPlanEvaluator( + state_sync, + mocker.Mock(), + mocker.Mock(), + None, + ) + + evaluator.visit_environment_record_update_stage(environment_stage, evaluatable_plan) + + _, kwargs = state_sync.promote.call_args + assert kwargs["max_trailing_gap_intervals"] == 1 + assert kwargs["planned_frontiers"] == {current.name: planned_frontier} + + +@pytest.mark.parametrize("invalid", [-1, True, 1.0, "1"]) +def test_max_trailing_gap_intervals_must_be_nonnegative(make_snapshot, invalid): + snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1"))) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + environment = Environment( + snapshots=[snapshot.table_info], + start_at="2024-01-01", + end_at="2024-01-03", + plan_id="plan", + ) + + with pytest.raises(ValidationError): + EvaluatablePlan( + start="2024-01-01", + end="2024-01-03", + new_snapshots=[snapshot], + environment=environment, + no_gaps=False, + skip_backfill=False, + empty_backfill=False, + restatements={}, + restate_all_snapshots=False, + is_dev=False, + allow_destructive_models=set(), + allow_additive_models=set(), + forward_only=False, + end_bounded=False, + ensure_finalized_snapshots=False, + directly_modified_snapshots=[], + indirectly_modified_snapshots={}, + metadata_updated_snapshots=[], + removed_snapshots=[], + requires_backfill=False, + disabled_restatement_models=set(), + max_trailing_gap_intervals=invalid, + ) diff --git a/tests/core/test_promotion.py b/tests/core/test_promotion.py new file mode 100644 index 0000000000..0d068f009a --- /dev/null +++ b/tests/core/test_promotion.py @@ -0,0 +1,480 @@ +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import pytest +from sqlglot import parse_one + +from sqlmesh.core.model import IncrementalByTimeRangeKind, SqlModel +from sqlmesh.core.node import IntervalUnit +from sqlmesh.core.promotion import ( + PromotionFrontier, + TrailingGapRejectionReason, + classify_trailing_gap, + evaluate_trailing_gap, +) +from sqlmesh.core.snapshot import Snapshot +from sqlmesh.utils.date import to_timestamp + + +def _snapshot_pair( + make_snapshot, + *, + cron: str = "@hourly", + allow_partials: bool = False, + interval_unit: IntervalUnit | None = None, + lookback: int = 0, + end: str | None = None, +) -> tuple[Snapshot, Snapshot]: + kind = IncrementalByTimeRangeKind( + time_column="ds", **({"lookback": lookback} if lookback else {}) + ) + model_kwargs = { + "name": "a", + "cron": cron, + "allow_partials": allow_partials, + "query": parse_one("select 1, ds"), + "kind": kind, + "start": "2023-01-01", + } + if interval_unit is not None: + model_kwargs["interval_unit"] = interval_unit + if end is not None: + model_kwargs["end"] = end + + incumbent_model = SqlModel(**model_kwargs) + target_model = incumbent_model.copy() + incumbent = make_snapshot(incumbent_model, version="incumbent") + target = make_snapshot(target_model, version="target") + target.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "target"}) + return incumbent, target + + +def _frontier(snapshot: Snapshot, end: str) -> PromotionFrontier: + return PromotionFrontier(snapshot_id=snapshot.snapshot_id, end=to_timestamp(end)) + + +def test_classify_accepts_one_stored_scheduler_suffix(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + result = classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + + assert result.start == to_timestamp("2023-01-01 11:00") + assert result.end == to_timestamp("2023-01-01 12:00") + assert result.interval_count == 1 + assert result.next_scheduler_boundary > to_timestamp("2023-01-01 12:00") + + +def test_classify_uses_stored_coverage_without_lookback_expansion(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot, lookback=14) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + result = classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + + assert result is not None + assert result.interval_count == 1 + assert result.interval == ( + to_timestamp("2023-01-01 11:00"), + to_timestamp("2023-01-01 12:00"), + ) + + +def test_classify_accepts_hourly_partial_suffix_inside_daily_interval(make_snapshot): + incumbent, target = _snapshot_pair( + make_snapshot, + allow_partials=True, + interval_unit=IntervalUnit.DAY, + ) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + result = classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + + assert result is not None + assert result.interval_count == 1 + + +@pytest.mark.parametrize( + "cron, interval_unit, planned_end, observed_end", + [ + ( + "0 */2 * * *", + None, + "2023-01-01 02:00", + "2023-01-01 04:00", + ), + ( + "@weekly", + None, + "2023-01-08 00:00", + "2023-01-15 00:00", + ), + ( + "@hourly", + IntervalUnit.HALF_HOUR, + "2023-01-01 01:00", + "2023-01-01 02:00", + ), + ], +) +def test_classify_counts_non_partial_scheduler_advances( + make_snapshot, cron, interval_unit, planned_end, observed_end +): + incumbent, target = _snapshot_pair(make_snapshot, cron=cron, interval_unit=interval_unit) + incumbent.add_interval("2023-01-01 00:00", observed_end) + target.add_interval("2023-01-01 00:00", planned_end) + + result = classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, planned_end), + 1, + start=to_timestamp("2023-01-01"), + ) + + assert result is not None + assert result.interval_count == 1 + + +@pytest.mark.parametrize( + "cron, planned_end, observed_end", + [ + ( + "0 */2 * * *", + "2023-01-01 02:00", + "2023-01-01 06:00", + ), + ( + "@weekly", + "2023-01-08 00:00", + "2023-01-22 00:00", + ), + ], +) +def test_classify_bounds_multiple_non_partial_scheduler_advances( + make_snapshot, cron, planned_end, observed_end +): + incumbent, target = _snapshot_pair(make_snapshot, cron=cron) + incumbent.add_interval("2023-01-01 00:00", observed_end) + target.add_interval("2023-01-01 00:00", planned_end) + frontier = _frontier(incumbent, planned_end) + start = to_timestamp("2023-01-01") + + assert classify_trailing_gap(target, incumbent, frontier, 1, start=start) is None + + result = classify_trailing_gap(target, incumbent, frontier, 2, start=start) + + assert result is not None + assert result.interval_count == 2 + + +def test_classify_preserves_floored_storage_boundaries_for_offset_cron(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot, cron="1 0 * * *") + incumbent.add_interval("2023-01-01 00:00", "2023-01-03 00:00") + target.add_interval("2023-01-01 00:00", "2023-01-02 00:00") + + result = classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-02 00:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + + assert result is not None + assert result.interval == ( + to_timestamp("2023-01-02 00:00"), + to_timestamp("2023-01-03 00:00"), + ) + assert result.interval_count == 1 + + +@pytest.mark.parametrize( + "planned_end, observed_end", + [ + ("2023-01-01 02:00", "2023-01-01 03:00"), + ("2023-01-01 03:00", "2023-01-01 04:00"), + ], +) +def test_classify_rejects_storage_aligned_non_scheduler_boundaries( + make_snapshot, planned_end, observed_end +): + incumbent, target = _snapshot_pair(make_snapshot, cron="0 */2 * * *") + incumbent.add_interval("2023-01-01 00:00", observed_end) + target.add_interval("2023-01-01 00:00", planned_end) + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, planned_end), + 2, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +@pytest.mark.parametrize( + "target_intervals, planned_end", + [ + ( + [ + ("2023-01-01 01:00", "2023-01-01 12:00"), + ], + "2023-01-01 11:00", + ), + ( + [ + ("2023-01-01 00:00", "2023-01-01 02:00"), + ("2023-01-01 03:00", "2023-01-01 05:00"), + ], + "2023-01-01 02:00", + ), + ], +) +def test_classify_rejects_historical_or_internal_holes( + make_snapshot, target_intervals, planned_end +): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 05:00") + for start, end in target_intervals: + target.add_interval(start, end) + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, planned_end), + 2, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +def test_classify_rejects_candidate_coverage_after_gap(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 04:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 02:00") + target.add_interval("2023-01-01 04:00", "2023-01-01 05:00") + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 02:00"), + 2, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +def test_classify_rejects_excessive_suffix(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 10:00") + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 10:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +def test_classify_rejects_suffix_past_target_model_end(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot, end="2023-01-01 11:00") + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +def test_classify_rejects_frontier_identity_or_schedule_mismatch(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + other_incumbent, _ = _snapshot_pair(make_snapshot) + other_incumbent.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "other"}) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + assert ( + classify_trailing_gap( + target, + incumbent, + _frontier(other_incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + changed_model = SqlModel( + name="a", + cron="@daily", + query=parse_one("select 1, ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + start="2023-01-01", + ) + changed_target = make_snapshot(changed_model, version="target") + changed_target.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "changed"}) + changed_target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + + assert ( + classify_trailing_gap( + changed_target, + incumbent, + _frontier(incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ) + is None + ) + + +def test_evaluate_reports_frontier_and_coverage_reasons(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 10:00") + frontier = _frontier(incumbent, "2023-01-01 11:00") + + assert ( + evaluate_trailing_gap( + target, + incumbent, + frontier, + 0, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.DISABLED + ) + assert ( + evaluate_trailing_gap( + target, + incumbent, + None, + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.FRONTIER_UNAVAILABLE + ) + other_incumbent, _ = _snapshot_pair(make_snapshot) + other_incumbent.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "other"}) + assert ( + evaluate_trailing_gap( + target, + incumbent, + _frontier(other_incumbent, "2023-01-01 11:00"), + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.FRONTIER_UNAVAILABLE + ) + assert ( + evaluate_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 12:00"), + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.FRONTIER_UNAVAILABLE + ) + assert ( + evaluate_trailing_gap( + target, + incumbent, + frontier, + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.PLANNED_COVERAGE_INCOMPLETE + ) + + +def test_evaluate_distinguishes_schedule_and_bound_rejections(make_snapshot): + incumbent, target = _snapshot_pair( + make_snapshot, + cron="0 */2 * * *", + ) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 03:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 02:00") + frontier = _frontier(incumbent, "2023-01-01 02:00") + + assert ( + evaluate_trailing_gap( + target, + incumbent, + frontier, + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.SCHEDULE_INELIGIBLE + ) + + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 13:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") + frontier = _frontier(incumbent, "2023-01-01 11:00") + assert ( + evaluate_trailing_gap( + target, + incumbent, + frontier, + 1, + start=to_timestamp("2023-01-01"), + ).reason + == TrailingGapRejectionReason.BOUND_EXCEEDED + ) + + +def test_evaluate_reports_non_trailing_gap(make_snapshot): + incumbent, target = _snapshot_pair(make_snapshot) + incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") + target.add_interval("2023-01-01 00:00", "2023-01-01 10:00") + target.add_interval("2023-01-01 11:00", "2023-01-01 12:00") + + result = evaluate_trailing_gap( + target, + incumbent, + _frontier(incumbent, "2023-01-01 10:00"), + 2, + start=to_timestamp("2023-01-01"), + ) + assert result.gap is None + assert result.reason == TrailingGapRejectionReason.NON_TRAILING_GAP diff --git a/tests/core/test_trailing_gap_integration.py b/tests/core/test_trailing_gap_integration.py new file mode 100644 index 0000000000..41c390d2c0 --- /dev/null +++ b/tests/core/test_trailing_gap_integration.py @@ -0,0 +1,224 @@ +# SPDX-License-Identifier: Apache-2.0 + +import pytest +import time_machine + +from sqlmesh.core.context import Context +from sqlmesh.core.plan.definition import EvaluatablePlan +from sqlmesh.core.snapshot import SnapshotChangeCategory +from sqlmesh.utils.date import to_timestamp +from sqlmesh.utils.errors import SQLMeshError + + +PLAN_TIME = "2024-01-01 04:00:00 UTC" +PROMOTION_TIME = "2024-01-01 05:00:00 UTC" +PLAN_FRONTIER = to_timestamp("2024-01-01 04:00") +OBSERVED_FRONTIER = to_timestamp("2024-01-01 05:00") + + +def _write_parent(path, value, *, partial_daily=False): + model_options = ( + """ + kind INCREMENTAL_BY_TIME_RANGE (time_column ts, batch_size 1, lookback 14), + cron '@hourly', + allow_partials true, + interval_unit 'day', +""" + if partial_daily + else """ + kind INCREMENTAL_BY_TIME_RANGE (time_column ts, batch_size 1), + cron '@hourly', +""" + ) + path.write_text( + f""" +MODEL ( + name test.parent, +{model_options} + start '2024-01-01' +); +SELECT ts, {value} AS value +FROM generate_series(@start_dt, @end_dt, INTERVAL '1 hour') AS t(ts) +""" + ) + + +def _create_project(tmp_path, *, partial_daily=False): + (tmp_path / "models").mkdir() + (tmp_path / "config.yaml").write_text("model_defaults:\n dialect: duckdb\n") + parent_path = tmp_path / "models" / "parent.sql" + _write_parent(parent_path, 1, partial_daily=partial_daily) + (tmp_path / "models" / "child.sql").write_text( + """ +MODEL ( + name test.child, + kind INCREMENTAL_BY_TIME_RANGE (time_column ts, batch_size 1), + cron '@hourly', + start '2024-01-01' +); +SELECT ts, value FROM test.parent WHERE ts >= @start_dt AND ts < @end_dt +""" + ) + return parent_path + + +def _add_interval_to_snapshots(context, snapshots, start, end): + for snapshot in snapshots: + if snapshot.is_model: + context.state_sync.add_interval(snapshot, start, end) + + +@pytest.mark.parametrize("bound", [0, 1]) +@pytest.mark.parametrize( + "race_variant,partial_daily", + [ + pytest.param("before_apply", False, id="before-apply-hourly"), + pytest.param("before_apply", True, id="before-apply-partial-daily-lookback"), + pytest.param("during_backfill", False, id="during-backfill-hourly"), + pytest.param("serialized", False, id="serialized-before-apply-hourly"), + ], +) +def test_apply_preserves_frontier_and_scheduler_repairs_lineage( + tmp_path, bound, race_variant, partial_daily +): + parent_path = _create_project(tmp_path, partial_daily=partial_daily) + + with time_machine.travel(PLAN_TIME, tick=False): + context = Context(paths=tmp_path) + context.apply(context.plan_builder("prod").build()) + incumbent = context.state_sync.get_environment("prod") + assert incumbent and incumbent.finalized_ts + _write_parent(parent_path, 2, partial_daily=partial_daily) + context.load() + builder = context.plan_builder("prod", no_gaps=True, max_trailing_gap_intervals=bound) + plan = builder.build() + for snapshot in plan.uncategorized: + builder.set_choice(snapshot, SnapshotChangeCategory.BREAKING) + plan = builder.build() + original_ends = dict(plan.end_override_per_model) + assert original_ends + assert all(to_timestamp(end) == PLAN_FRONTIER for end in original_ends.values()) + + serialized_plan = ( + EvaluatablePlan.parse_raw(plan.to_evaluatable().json()) + if race_variant == "serialized" + else None + ) + if serialized_plan: + assert serialized_plan.max_trailing_gap_intervals == bound + assert serialized_plan.end_override_per_model == original_ends + assert serialized_plan.planned_frontiers == plan.planned_frontiers + + with time_machine.travel(PROMOTION_TIME, tick=False): + if race_variant in {"before_apply", "serialized"}: + assert context.run() + assert all( + end == OBSERVED_FRONTIER + for end in context.state_sync.max_interval_end_per_model("prod").values() + ) + + if race_variant == "during_backfill": + real_promote = context.state_sync.promote + + def promote_with_race(*args, **kwargs): + candidate_snapshots = context.state_sync.get_snapshots( + plan.environment.snapshots + ).values() + assert all( + snapshot.intervals + and max(end for _, end in snapshot.intervals) == PLAN_FRONTIER + for snapshot in candidate_snapshots + if snapshot.is_model + ) + _add_interval_to_snapshots( + context, + context.state_sync.get_snapshots(incumbent.snapshots).values(), + PLAN_FRONTIER, + OBSERVED_FRONTIER, + ) + return real_promote(*args, **kwargs) + + context.state_sync.promote = promote_with_race + + def evaluate_plan(): + if serialized_plan: + context._scheduler.create_plan_evaluator(context).evaluate(serialized_plan) + else: + context.apply(plan) + + if not bound: + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + evaluate_plan() + assert context.state_sync.get_environment("prod").plan_id == incumbent.plan_id + context.close() + return + + assert all(f.end == PLAN_FRONTIER for f in plan.planned_frontiers.values()) + assert plan.planned_frontiers + evaluate_plan() + assert plan.end_override_per_model == original_ends + if serialized_plan: + assert serialized_plan.end_override_per_model == original_ends + promoted = context.state_sync.get_environment("prod") + assert promoted.plan_id == plan.plan_id + assert promoted.finalized_ts + assert all( + end == PLAN_FRONTIER + for end in context.state_sync.max_interval_end_per_model("prod").values() + ) + assert context.engine_adapter.fetchall( + "SELECT count(*), min(value), max(value) FROM test.child" + ) == [(4, 2, 2)] + assert context.run() + assert all( + end == OBSERVED_FRONTIER + for end in context.state_sync.max_interval_end_per_model("prod").values() + ) + assert context.engine_adapter.fetchall( + "SELECT count(*), min(value), max(value) FROM test.child" + ) == [(5, 2, 2)] + context.close() + + +def test_apply_rejects_trailing_gap_with_downstream_coverage(tmp_path): + parent_path = _create_project(tmp_path) + + with time_machine.travel(PLAN_TIME, tick=False): + context = Context(paths=tmp_path) + context.apply(context.plan_builder("prod").build()) + incumbent = context.state_sync.get_environment("prod") + assert incumbent and incumbent.finalized_ts + _write_parent(parent_path, 2) + context.load() + builder = context.plan_builder("prod", no_gaps=True, max_trailing_gap_intervals=1) + plan = builder.build() + for snapshot in plan.uncategorized: + builder.set_choice(snapshot, SnapshotChangeCategory.BREAKING) + plan = builder.build() + + with time_machine.travel(PROMOTION_TIME, tick=False): + real_promote = context.state_sync.promote + + def promote_with_downstream_coverage(*args, **kwargs): + _add_interval_to_snapshots( + context, + context.state_sync.get_snapshots(incumbent.snapshots).values(), + PLAN_FRONTIER, + OBSERVED_FRONTIER, + ) + candidate_snapshots = context.state_sync.get_snapshots( + plan.environment.snapshots + ).values() + child = next( + snapshot + for snapshot in candidate_snapshots + if snapshot.node.name == context.get_model("test.child").name + ) + context.state_sync.add_interval(child, PLAN_FRONTIER, OBSERVED_FRONTIER) + return real_promote(*args, **kwargs) + + context.state_sync.promote = promote_with_downstream_coverage + with pytest.raises(SQLMeshError, match="Detected missing intervals"): + context.apply(plan) + assert context.state_sync.get_environment("prod").plan_id == incumbent.plan_id + context.close() diff --git a/tests/integrations/github/cicd/test_config.py b/tests/integrations/github/cicd/test_config.py index 20b72393dc..1ccc3149fa 100644 --- a/tests/integrations/github/cicd/test_config.py +++ b/tests/integrations/github/cicd/test_config.py @@ -1,3 +1,4 @@ +import math import pathlib import pytest @@ -9,7 +10,7 @@ load_config_from_paths, ) from sqlmesh.utils.errors import ConfigError -from sqlmesh.integrations.github.cicd.config import MergeMethod +from sqlmesh.integrations.github.cicd.config import GithubCICDBotConfig, MergeMethod from tests.utils.test_filesystem import create_temp_file pytestmark = pytest.mark.github @@ -44,6 +45,7 @@ def test_load_yaml_config_default(tmp_path): assert config.cicd_bot.prod_branch_names == ["main", "master"] assert not config.cicd_bot.pr_min_intervals assert config.cicd_bot.pr_preview_min_intervals == 1 + assert config.cicd_bot.max_trailing_gap_intervals_on_deploy_to_prod == 0 def test_load_yaml_config(tmp_path): @@ -70,6 +72,7 @@ def test_load_yaml_config(tmp_path): prod_branch_name: testing pr_min_intervals: 1 pr_preview_min_intervals: 2 + max_trailing_gap_intervals_on_deploy_to_prod: 2 model_defaults: dialect: duckdb """, @@ -97,6 +100,7 @@ def test_load_yaml_config(tmp_path): assert config.cicd_bot.prod_branch_names == ["testing"] assert config.cicd_bot.pr_min_intervals == 1 assert config.cicd_bot.pr_preview_min_intervals == 2 + assert config.cicd_bot.max_trailing_gap_intervals_on_deploy_to_prod == 2 def test_load_python_config_defaults(tmp_path): @@ -131,6 +135,7 @@ def test_load_python_config_defaults(tmp_path): assert config.cicd_bot.prod_branch_names == ["main", "master"] assert not config.cicd_bot.pr_min_intervals assert config.cicd_bot.pr_preview_min_intervals == 1 + assert config.cicd_bot.max_trailing_gap_intervals_on_deploy_to_prod == 0 def test_load_python_config(tmp_path): @@ -227,6 +232,12 @@ def test_validation(tmp_path): load_config_from_paths(Config, project_paths=[tmp_path / "config.yaml"]) +@pytest.mark.parametrize("value", [-1, 1.5, math.inf, math.nan, True, "1"]) +def test_max_trailing_gap_intervals_on_deploy_to_prod_requires_nonnegative_finite_int(value): + with pytest.raises(ValueError): + GithubCICDBotConfig(max_trailing_gap_intervals_on_deploy_to_prod=value) + + def test_ttl_in_past(tmp_path): create_temp_file( tmp_path, diff --git a/tests/integrations/github/cicd/test_github_controller.py b/tests/integrations/github/cicd/test_github_controller.py index dbb09f4336..cd0e7401b3 100644 --- a/tests/integrations/github/cicd/test_github_controller.py +++ b/tests/integrations/github/cicd/test_github_controller.py @@ -378,6 +378,45 @@ def test_prod_plan(github_client, make_controller): ) +def test_prod_plan_forwards_max_trailing_gap_intervals( + github_client, make_controller, mocker: MockerFixture +): + context = mocker.MagicMock() + plan_builder = mocker.MagicMock() + plan = mocker.MagicMock(spec=Plan) + plan_builder.build.return_value = plan + context.config.plan.forward_only = False + context.plan_builder.return_value = plan_builder + mocker.patch("sqlmesh.integrations.github.cicd.controller.Context", return_value=context) + + bot_config = GithubCICDBotConfig(max_trailing_gap_intervals_on_deploy_to_prod=2) + controller = make_controller( + "tests/fixtures/github/pull_request_synchronized.json", + github_client, + bot_config=bot_config, + ) + + assert controller.prod_plan is plan + context.plan_builder.assert_called_once_with( + c.PROD, + no_gaps=True, + skip_tests=True, + skip_linter=True, + categorizer_config=bot_config.auto_categorize_changes, + run=bot_config.run_on_deploy_to_prod, + forward_only=controller.forward_only_plan, + max_trailing_gap_intervals=2, + ) + + context.plan_builder.reset_mock() + plan_builder.build.reset_mock() + assert controller.prod_plan_with_gaps is plan + prod_plan_with_gaps_call = context.plan_builder.call_args + assert prod_plan_with_gaps_call.args == (c.PROD,) + assert prod_plan_with_gaps_call.kwargs["no_gaps"] is False + assert "max_trailing_gap_intervals" not in prod_plan_with_gaps_call.kwargs + + def test_prod_plan_auto_categorization(github_client, make_controller): custom_categorizer_config = CategorizerConfig.all_off() default_pr_start = "1 week ago" From 7be4da250e3ef4886309d2aef966b127a4ad6161 Mon Sep 17 00:00:00 2001 From: Jesse Hodges Date: Wed, 21 Oct 2026 18:07:25 -0500 Subject: [PATCH 2/3] fix: harden trailing-gap promotion diagnostics and simplify policy checks Signed-off-by: Jesse Hodges --- sqlmesh/core/promotion.py | 50 +++---------- sqlmesh/core/state_sync/db/facade.py | 95 +++++++++++++++++------- tests/core/state_sync/test_state_sync.py | 47 +++++++++++- tests/core/test_promotion.py | 77 ++++++++++--------- 4 files changed, 161 insertions(+), 108 deletions(-) diff --git a/sqlmesh/core/promotion.py b/sqlmesh/core/promotion.py index 3893944e2c..d388e8ae90 100644 --- a/sqlmesh/core/promotion.py +++ b/sqlmesh/core/promotion.py @@ -59,26 +59,24 @@ class TrailingGapEvaluation: reason: t.Optional[TrailingGapRejectionReason] = None -def classify_trailing_gap( +def evaluate_trailing_gap( target_snapshot: Snapshot, incumbent_snapshot: Snapshot, planned_frontier: t.Optional[PromotionFrontier], max_trailing_gap_intervals: int, *, start: int, -) -> t.Optional[TrailingGap]: - """Classify an absent suffix using stored snapshot coverage. +) -> TrailingGapEvaluation: + """Evaluate whether an absent suffix is a bounded trailing gap. - The existing promotion check deliberately asks ``Snapshot.missing_intervals`` - for the model's required work. That method expands missing work for - lookback, so its result cannot be used as a safety-policy count. This - helper instead subtracts the candidate's *stored* intervals from the - incumbent's stored intervals and then counts the resulting suffix against - the model's scheduler cron boundaries. + This subtracts the candidate's *stored* intervals from the incumbent's + stored intervals. ``Snapshot.missing_intervals`` cannot provide a safe + policy count because it expands work for lookback. The resulting suffix is + counted against the model's scheduler cron boundaries. - ``None`` means that the change cannot be proven to be a bounded trailing - suffix. Callers should retain the normal strict no-gap failure in that - case. + ``gap`` is populated only when the suffix is accepted. Otherwise ``reason`` + gives a stable category for the policy rejection. Callers should retain the + normal strict no-gap failure when ``gap`` is absent. Args: target_snapshot: The candidate snapshot being promoted. @@ -88,34 +86,10 @@ def classify_trailing_gap( plan's original backfill bounds were selected. max_trailing_gap_intervals: Maximum number of scheduler boundaries the accepted suffix may span. - start: Inclusive start of the original no-gap obligation. The + start: Inclusive start of the original no-gap obligation. The promotion facade passes this value so historical coverage is checked independently of lookback. """ - return evaluate_trailing_gap( - target_snapshot, - incumbent_snapshot, - planned_frontier, - max_trailing_gap_intervals, - start=start, - ).gap - - -def evaluate_trailing_gap( - target_snapshot: Snapshot, - incumbent_snapshot: Snapshot, - planned_frontier: t.Optional[PromotionFrontier], - max_trailing_gap_intervals: int, - *, - start: int, -) -> TrailingGapEvaluation: - """Evaluate an absent suffix and return its bounded-policy reason. - - :func:`classify_trailing_gap` is a convenience wrapper that returns only - the optional accepted gap. Callers that report policy decisions should use - this function so the policy is evaluated exactly once and a rejected gap - is assigned a bounded, stable category. - """ if max_trailing_gap_intervals < 1: return _rejected(TrailingGapRejectionReason.DISABLED) @@ -187,8 +161,6 @@ def evaluate_trailing_gap( boundaries = boundary_evaluation.boundaries if boundaries is None or not boundaries: return _rejected(TrailingGapRejectionReason.SCHEDULE_INELIGIBLE) - if len(boundaries) > max_trailing_gap_intervals: - return _rejected(TrailingGapRejectionReason.BOUND_EXCEEDED) try: # The warning describes the next scheduler run from the current wall diff --git a/sqlmesh/core/state_sync/db/facade.py b/sqlmesh/core/state_sync/db/facade.py index b3f393b2d0..9732bde610 100644 --- a/sqlmesh/core/state_sync/db/facade.py +++ b/sqlmesh/core/state_sync/db/facade.py @@ -266,26 +266,12 @@ def promote( ) for target_snapshot, incumbent_snapshot, frontier, gap in accepted_trailing_gaps: - payload = { - "environment": environment.name, - "plan_id": environment.plan_id, - "previous_plan_id": environment.previous_plan_id, - "model": target_snapshot.name, - "incumbent_snapshot_id": str(incumbent_snapshot.snapshot_id), - "candidate_snapshot_id": str(target_snapshot.snapshot_id), - "planned_frontier": frontier.end, - "observed_frontier": gap.end, - "gap_count": gap.interval_count, - "gap_range": gap.interval, - "next_scheduler_boundary": gap.next_scheduler_boundary, - } - logger.warning( - "Accepted bounded trailing gap during promotion: %s", - payload, - extra={ - "sqlmesh_event": "bounded_trailing_gap_accepted", - "sqlmesh_promotion": payload, - }, + _log_trailing_gap_acceptance( + environment, + target_snapshot, + incumbent_snapshot, + frontier, + gap, ) return result @@ -682,9 +668,14 @@ def _ensure_no_gaps( max_trailing_gap_intervals, start=start, ) - if evaluation.gap and frontier: + if evaluation.gap is not None: accepted_trailing_gaps.append( - (target_snapshot, prev_snapshot, frontier, evaluation.gap) + ( + target_snapshot, + prev_snapshot, + t.cast(PromotionFrontier, frontier), + evaluation.gap, + ) ) continue @@ -700,11 +691,15 @@ def _ensure_no_gaps( "Please re-apply your plan to resolve this error." ) + children_by_parent = ( + _build_children_by_parent(target_snapshots_by_name) if accepted_trailing_gaps else {} + ) for target_snapshot, _, _, trailing_gap in accepted_trailing_gaps: if _has_downstream_coverage( target_snapshots_by_name, target_snapshot.name, trailing_gap, + children_by_parent, ): _log_trailing_gap_rejection( target_environment, @@ -778,8 +773,57 @@ def _log_trailing_gap_rejection( pass +def _log_trailing_gap_acceptance( + environment: Environment, + target_snapshot: Snapshot, + incumbent_snapshot: Snapshot, + frontier: PromotionFrontier, + gap: TrailingGap, +) -> None: + """Report an accepted policy decision without affecting promotion.""" + try: + payload = { + "environment": environment.name, + "plan_id": environment.plan_id, + "previous_plan_id": environment.previous_plan_id, + "model": target_snapshot.name, + "incumbent_snapshot_id": str(incumbent_snapshot.snapshot_id), + "candidate_snapshot_id": str(target_snapshot.snapshot_id), + "planned_frontier": frontier.end, + "observed_frontier": gap.end, + "gap_count": gap.interval_count, + "gap_range": gap.interval, + "next_scheduler_boundary": gap.next_scheduler_boundary, + } + logger.warning( + "Accepted bounded trailing gap during promotion: %s", + payload, + extra={ + "sqlmesh_event": "bounded_trailing_gap_accepted", + "sqlmesh_promotion": payload, + }, + ) + except Exception: + # Observability must not roll back a successful promotion. + pass + + +def _build_children_by_parent( + snapshots_by_name: t.Mapping[str, Snapshot], +) -> t.Dict[str, t.Set[str]]: + """Build target snapshot adjacency for a single no-gaps evaluation.""" + children_by_parent: t.Dict[str, t.Set[str]] = {} + for snapshot in snapshots_by_name.values(): + for parent in snapshot.parents: + children_by_parent.setdefault(parent.name, set()).add(snapshot.name) + return children_by_parent + + def _has_downstream_coverage( - snapshots_by_name: t.Mapping[str, Snapshot], model_name: str, gap: TrailingGap + snapshots_by_name: t.Mapping[str, Snapshot], + model_name: str, + gap: TrailingGap, + children_by_parent: t.Mapping[str, t.Set[str]], ) -> bool: """Return whether a target descendant already covers an accepted parent tail. @@ -788,11 +832,6 @@ def _has_downstream_coverage( conservative avoids accepting that state when auto-upstream selection or a later scheduler run cannot invalidate the descendant interval. """ - children_by_parent: t.Dict[str, t.Set[str]] = {} - for snapshot in snapshots_by_name.values(): - for parent in snapshot.parents: - children_by_parent.setdefault(parent.name, set()).add(snapshot.name) - descendants: t.Set[str] = set() pending = [model_name] while pending: diff --git a/tests/core/state_sync/test_state_sync.py b/tests/core/state_sync/test_state_sync.py index 6d684b79a3..61dbe87511 100644 --- a/tests/core/state_sync/test_state_sync.py +++ b/tests/core/state_sync/test_state_sync.py @@ -2,7 +2,7 @@ import logging import re import typing as t -from unittest.mock import call, patch +from unittest.mock import PropertyMock, call, patch import duckdb # noqa: TID253 import pandas as pd # noqa: TID253 @@ -24,7 +24,7 @@ SeedModel, SqlModel, ) -from sqlmesh.core.promotion import PromotionFrontier +from sqlmesh.core.promotion import PromotionFrontier, TrailingGap from sqlmesh.core.snapshot import ( Snapshot, SnapshotChangeCategory, @@ -1253,6 +1253,49 @@ def test_promote_snapshots_accepts_bounded_trailing_gap( assert payload["gap_count"] == 1 +@pytest.mark.parametrize("failure", ["warning", "payload"]) +def test_promote_snapshots_acceptance_logging_failures_do_not_block_promotion( + state_sync: EngineAdapterStateSync, + make_snapshot: t.Callable, + failure: str, +): + _, target, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + logging_patch: t.ContextManager[t.Any] + if failure == "warning": + logging_patch = patch( + "sqlmesh.core.state_sync.db.facade.logger.warning", + side_effect=RuntimeError("logging failed"), + ) + else: + logging_patch = patch.object( + TrailingGap, + "interval", + new_callable=PropertyMock, + side_effect=RuntimeError("payload construction failed"), + ) + + with logging_patch: + result = promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + assert result.added == [target.table_info] + environment = state_sync.get_environment("prod") + assert environment is not None + assert environment.plan_id == "new_plan_id" + assert environment.snapshots[0].snapshot_id == target.snapshot_id + assert state_sync.get_snapshots([target])[target.snapshot_id].intervals == [ + (to_timestamp("2023-01-01 00:00"), to_timestamp("2023-01-01 11:00")) + ] + + def test_promote_snapshots_rolls_back_multiple_models_before_warning( state_sync: EngineAdapterStateSync, make_snapshot: t.Callable, diff --git a/tests/core/test_promotion.py b/tests/core/test_promotion.py index 0d068f009a..1a5f7765ff 100644 --- a/tests/core/test_promotion.py +++ b/tests/core/test_promotion.py @@ -10,7 +10,6 @@ from sqlmesh.core.promotion import ( PromotionFrontier, TrailingGapRejectionReason, - classify_trailing_gap, evaluate_trailing_gap, ) from sqlmesh.core.snapshot import Snapshot @@ -54,18 +53,18 @@ def _frontier(snapshot: Snapshot, end: str) -> PromotionFrontier: return PromotionFrontier(snapshot_id=snapshot.snapshot_id, end=to_timestamp(end)) -def test_classify_accepts_one_stored_scheduler_suffix(make_snapshot): +def test_evaluate_accepts_one_stored_scheduler_suffix(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot) incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") - result = classify_trailing_gap( + result = evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap assert result.start == to_timestamp("2023-01-01 11:00") assert result.end == to_timestamp("2023-01-01 12:00") @@ -73,18 +72,18 @@ def test_classify_accepts_one_stored_scheduler_suffix(make_snapshot): assert result.next_scheduler_boundary > to_timestamp("2023-01-01 12:00") -def test_classify_uses_stored_coverage_without_lookback_expansion(make_snapshot): +def test_evaluate_uses_stored_coverage_without_lookback_expansion(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot, lookback=14) incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") - result = classify_trailing_gap( + result = evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap assert result is not None assert result.interval_count == 1 @@ -94,7 +93,7 @@ def test_classify_uses_stored_coverage_without_lookback_expansion(make_snapshot) ) -def test_classify_accepts_hourly_partial_suffix_inside_daily_interval(make_snapshot): +def test_evaluate_accepts_hourly_partial_suffix_inside_daily_interval(make_snapshot): incumbent, target = _snapshot_pair( make_snapshot, allow_partials=True, @@ -103,13 +102,13 @@ def test_classify_accepts_hourly_partial_suffix_inside_daily_interval(make_snaps incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") - result = classify_trailing_gap( + result = evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap assert result is not None assert result.interval_count == 1 @@ -138,20 +137,20 @@ def test_classify_accepts_hourly_partial_suffix_inside_daily_interval(make_snaps ), ], ) -def test_classify_counts_non_partial_scheduler_advances( +def test_evaluate_counts_non_partial_scheduler_advances( make_snapshot, cron, interval_unit, planned_end, observed_end ): incumbent, target = _snapshot_pair(make_snapshot, cron=cron, interval_unit=interval_unit) incumbent.add_interval("2023-01-01 00:00", observed_end) target.add_interval("2023-01-01 00:00", planned_end) - result = classify_trailing_gap( + result = evaluate_trailing_gap( target, incumbent, _frontier(incumbent, planned_end), 1, start=to_timestamp("2023-01-01"), - ) + ).gap assert result is not None assert result.interval_count == 1 @@ -172,7 +171,7 @@ def test_classify_counts_non_partial_scheduler_advances( ), ], ) -def test_classify_bounds_multiple_non_partial_scheduler_advances( +def test_evaluate_bounds_multiple_non_partial_scheduler_advances( make_snapshot, cron, planned_end, observed_end ): incumbent, target = _snapshot_pair(make_snapshot, cron=cron) @@ -181,26 +180,26 @@ def test_classify_bounds_multiple_non_partial_scheduler_advances( frontier = _frontier(incumbent, planned_end) start = to_timestamp("2023-01-01") - assert classify_trailing_gap(target, incumbent, frontier, 1, start=start) is None + assert evaluate_trailing_gap(target, incumbent, frontier, 1, start=start).gap is None - result = classify_trailing_gap(target, incumbent, frontier, 2, start=start) + result = evaluate_trailing_gap(target, incumbent, frontier, 2, start=start).gap assert result is not None assert result.interval_count == 2 -def test_classify_preserves_floored_storage_boundaries_for_offset_cron(make_snapshot): +def test_evaluate_preserves_floored_storage_boundaries_for_offset_cron(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot, cron="1 0 * * *") incumbent.add_interval("2023-01-01 00:00", "2023-01-03 00:00") target.add_interval("2023-01-01 00:00", "2023-01-02 00:00") - result = classify_trailing_gap( + result = evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-02 00:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap assert result is not None assert result.interval == ( @@ -217,7 +216,7 @@ def test_classify_preserves_floored_storage_boundaries_for_offset_cron(make_snap ("2023-01-01 03:00", "2023-01-01 04:00"), ], ) -def test_classify_rejects_storage_aligned_non_scheduler_boundaries( +def test_evaluate_rejects_storage_aligned_non_scheduler_boundaries( make_snapshot, planned_end, observed_end ): incumbent, target = _snapshot_pair(make_snapshot, cron="0 */2 * * *") @@ -225,13 +224,13 @@ def test_classify_rejects_storage_aligned_non_scheduler_boundaries( target.add_interval("2023-01-01 00:00", planned_end) assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(incumbent, planned_end), 2, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) @@ -254,7 +253,7 @@ def test_classify_rejects_storage_aligned_non_scheduler_boundaries( ), ], ) -def test_classify_rejects_historical_or_internal_holes( +def test_evaluate_rejects_historical_or_internal_holes( make_snapshot, target_intervals, planned_end ): incumbent, target = _snapshot_pair(make_snapshot) @@ -263,70 +262,70 @@ def test_classify_rejects_historical_or_internal_holes( target.add_interval(start, end) assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(incumbent, planned_end), 2, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) -def test_classify_rejects_candidate_coverage_after_gap(make_snapshot): +def test_evaluate_rejects_candidate_coverage_after_gap(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot) incumbent.add_interval("2023-01-01 00:00", "2023-01-01 04:00") target.add_interval("2023-01-01 00:00", "2023-01-01 02:00") target.add_interval("2023-01-01 04:00", "2023-01-01 05:00") assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 02:00"), 2, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) -def test_classify_rejects_excessive_suffix(make_snapshot): +def test_evaluate_rejects_excessive_suffix(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot) incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") target.add_interval("2023-01-01 00:00", "2023-01-01 10:00") assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 10:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) -def test_classify_rejects_suffix_past_target_model_end(make_snapshot): +def test_evaluate_rejects_suffix_past_target_model_end(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot, end="2023-01-01 11:00") incumbent.add_interval("2023-01-01 00:00", "2023-01-01 12:00") target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) -def test_classify_rejects_frontier_identity_or_schedule_mismatch(make_snapshot): +def test_evaluate_rejects_frontier_identity_or_schedule_mismatch(make_snapshot): incumbent, target = _snapshot_pair(make_snapshot) other_incumbent, _ = _snapshot_pair(make_snapshot) other_incumbent.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "other"}) @@ -334,13 +333,13 @@ def test_classify_rejects_frontier_identity_or_schedule_mismatch(make_snapshot): target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") assert ( - classify_trailing_gap( + evaluate_trailing_gap( target, incumbent, _frontier(other_incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) @@ -356,13 +355,13 @@ def test_classify_rejects_frontier_identity_or_schedule_mismatch(make_snapshot): changed_target.add_interval("2023-01-01 00:00", "2023-01-01 11:00") assert ( - classify_trailing_gap( + evaluate_trailing_gap( changed_target, incumbent, _frontier(incumbent, "2023-01-01 11:00"), 1, start=to_timestamp("2023-01-01"), - ) + ).gap is None ) From 2ac56207c725e8e42c651c6890cfbcf633e15d49 Mon Sep 17 00:00:00 2001 From: Jesse Hodges Date: Wed, 21 Oct 2026 20:43:15 -0500 Subject: [PATCH 3/3] fix: explain recovery for trailing-gap policy rejections Signed-off-by: Jesse Hodges --- sqlmesh/core/state_sync/db/facade.py | 50 +++++++++++++++--- tests/core/state_sync/test_state_sync.py | 66 +++++++++++++++++++++--- 2 files changed, 102 insertions(+), 14 deletions(-) diff --git a/sqlmesh/core/state_sync/db/facade.py b/sqlmesh/core/state_sync/db/facade.py index 9732bde610..44bbfede9c 100644 --- a/sqlmesh/core/state_sync/db/facade.py +++ b/sqlmesh/core/state_sync/db/facade.py @@ -686,10 +686,7 @@ def _ensure_no_gaps( candidate_plan_id=candidate_plan_id, ) - raise SQLMeshError( - f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " - "Please re-apply your plan to resolve this error." - ) + raise _missing_intervals_error(target_snapshot, evaluation.reason) children_by_parent = ( _build_children_by_parent(target_snapshots_by_name) if accepted_trailing_gaps else {} @@ -707,9 +704,8 @@ def _ensure_no_gaps( TrailingGapRejectionReason.DOWNSTREAM_COVERAGE, candidate_plan_id=candidate_plan_id, ) - raise SQLMeshError( - f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " - "Please re-apply your plan to resolve this error." + raise _missing_intervals_error( + target_snapshot, TrailingGapRejectionReason.DOWNSTREAM_COVERAGE ) return accepted_trailing_gaps @@ -744,6 +740,46 @@ def _remove_partial_intervals( return results +def _missing_intervals_error( + target_snapshot: Snapshot, + reason: t.Optional[TrailingGapRejectionReason], +) -> SQLMeshError: + prefix = f"Detected missing intervals for model {target_snapshot.name}, interrupting your current plan. " + if reason is None or reason == TrailingGapRejectionReason.DISABLED: + return SQLMeshError(prefix + "Please re-apply your plan to resolve this error.") + + guidance = { + TrailingGapRejectionReason.BOUND_EXCEEDED: ( + "Backfill the candidate's missing coverage through the incumbent frontier before retrying promotion. " + "Alternatively, review whether a larger allowance is appropriate and create a fresh plan; " + "changing the allowance does not change an existing serialized plan." + ), + TrailingGapRejectionReason.DOWNSTREAM_COVERAGE: ( + "Backfill the candidate's upstream gap, invalidate overlapping downstream coverage, and then recompute it " + "before retrying promotion." + ), + TrailingGapRejectionReason.SCHEDULE_INELIGIBLE: ( + "Resolve the incompatible schedule or interval semantics and backfill the candidate's missing " + "coverage before retrying promotion; increasing the allowance cannot make this gap eligible." + ), + TrailingGapRejectionReason.FRONTIER_UNAVAILABLE: ( + "Create a fresh plan that captures the current incumbent frontier and backfills the candidate's " + "required coverage; re-applying this serialized plan will not refresh its frontier or bounds." + ), + TrailingGapRejectionReason.PLANNED_COVERAGE_INCOMPLETE: ( + "Backfill the candidate's missing required coverage before retrying promotion. The allowance only " + "applies to a new trailing suffix." + ), + TrailingGapRejectionReason.NON_TRAILING_GAP: ( + "Backfill the candidate's missing historical or internal coverage before retrying promotion. " + "The allowance only applies to a new trailing suffix." + ), + }.get(reason, "Inspect and repair the candidate's missing coverage before retrying promotion.") + return SQLMeshError( + f"{prefix}Bounded trailing-gap policy rejected ({reason.value}). {guidance}" + ) + + def _log_trailing_gap_rejection( target_environment: Environment, target_snapshot: Snapshot, diff --git a/tests/core/state_sync/test_state_sync.py b/tests/core/state_sync/test_state_sync.py index 61dbe87511..6873640e69 100644 --- a/tests/core/state_sync/test_state_sync.py +++ b/tests/core/state_sync/test_state_sync.py @@ -1156,11 +1156,12 @@ def test_promote_snapshots_no_gaps(state_sync: EngineAdapterStateSync, make_snap ) state_sync.push_snapshots([new_snapshot_missing_interval]) state_sync.add_interval(new_snapshot_missing_interval, "2022-01-01", "2022-01-02") - with pytest.raises( - SQLMeshError, - match=r".*Detected missing intervals for model .*, interrupting your current plan. Please re-apply your plan to resolve this error.*", - ): + with pytest.raises(SQLMeshError) as error: promote_snapshots(state_sync, [new_snapshot_missing_interval], "prod", no_gaps=True) + assert str(error.value) == ( + f"Detected missing intervals for model {new_snapshot_missing_interval.name}, " + "interrupting your current plan. Please re-apply your plan to resolve this error." + ) new_snapshot_same_interval = make_snapshot(model, version="c") new_snapshot_same_interval.change_category = SnapshotChangeCategory.BREAKING @@ -1307,7 +1308,7 @@ def test_promote_snapshots_rolls_back_multiple_models_before_warning( caplog.clear() with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): - with pytest.raises(SQLMeshError, match="Detected missing intervals"): + with pytest.raises(SQLMeshError, match="Detected missing intervals") as error: promote_snapshots( state_sync, targets, @@ -1318,6 +1319,15 @@ def test_promote_snapshots_rolls_back_multiple_models_before_warning( plan_id="new_plan_id", previous_plan_id="test_plan_id", ) + assert "(bound_exceeded)" in str(error.value) + assert "Backfill the candidate's missing coverage through the incumbent frontier" in str( + error.value + ) + assert "review whether a larger allowance is appropriate and create a fresh plan" in str( + error.value + ) + assert "changing the allowance does not change an existing serialized plan" in str(error.value) + assert "Please re-apply" not in str(error.value) assert not any( getattr(record, "sqlmesh_event", None) == "bounded_trailing_gap_accepted" @@ -1437,7 +1447,7 @@ def test_promote_snapshots_reports_downstream_policy_rejection( ) with caplog.at_level(logging.WARNING, logger="sqlmesh.core.state_sync.db.facade"): - with pytest.raises(SQLMeshError, match="Detected missing intervals"): + with pytest.raises(SQLMeshError, match="Detected missing intervals") as error: promote_snapshots( state_sync, target_snapshots, @@ -1448,6 +1458,9 @@ def test_promote_snapshots_reports_downstream_policy_rejection( plan_id="candidate_plan_id", previous_plan_id="test_plan_id", ) + assert "(downstream_coverage)" in str(error.value) + assert "invalidate overlapping downstream coverage, and then recompute it" in str(error.value) + assert "Please re-apply" not in str(error.value) rejected_warnings = [ record @@ -1460,6 +1473,41 @@ def test_promote_snapshots_reports_downstream_policy_rejection( assert payload["plan_id"] == "candidate_plan_id" +def test_promote_snapshots_gives_actionable_schedule_rejection( + state_sync: EngineAdapterStateSync, make_snapshot: t.Callable +): + incumbent, _, frontier = _prepare_trailing_gap_promotion(state_sync, make_snapshot) + daily_model = SqlModel( + name=incumbent.name, + query=parse_one("select 1, ds"), + kind=IncrementalByTimeRangeKind(time_column="ds"), + cron="@daily", + start="2023-01-01", + ) + target = make_snapshot(daily_model, version="daily_target") + target.change_category = SnapshotChangeCategory.BREAKING + target.fingerprint = incumbent.fingerprint.copy(update={"data_hash": "daily_target"}) + state_sync.push_snapshots([target]) + state_sync.add_interval(incumbent, "2023-01-01 12:00", "2023-01-02 00:00") + + with pytest.raises(SQLMeshError, match="Detected missing intervals") as error: + promote_snapshots( + state_sync, + [target], + "prod", + no_gaps=True, + max_trailing_gap_intervals=1, + planned_frontiers={target.name: frontier}, + plan_id="new_plan_id", + previous_plan_id="test_plan_id", + ) + + assert "(schedule_ineligible)" in str(error.value) + assert "Resolve the incompatible schedule or interval semantics" in str(error.value) + assert "increasing the allowance cannot make this gap eligible" in str(error.value) + assert "Please re-apply" not in str(error.value) + + def test_promote_snapshots_rejects_frontier_identity_mismatch( state_sync: EngineAdapterStateSync, make_snapshot: t.Callable ): @@ -1469,7 +1517,7 @@ def test_promote_snapshots_rejects_frontier_identity_mismatch( end=frontier.end, ) - with pytest.raises(SQLMeshError, match="Detected missing intervals"): + with pytest.raises(SQLMeshError, match="Detected missing intervals") as error: promote_snapshots( state_sync, [target], @@ -1480,6 +1528,10 @@ def test_promote_snapshots_rejects_frontier_identity_mismatch( plan_id="new_plan_id", previous_plan_id="test_plan_id", ) + assert "(frontier_unavailable)" in str(error.value) + assert "Create a fresh plan that captures the current incumbent frontier" in str(error.value) + assert "backfills the candidate's required coverage" in str(error.value) + assert "Please re-apply" not in str(error.value) def test_promote_snapshots_conflict_precedes_relaxed_gap_check(