diff --git a/docs/concepts/plans.md b/docs/concepts/plans.md index c3e12652ee..0ec5c4e315 100644 --- a/docs/concepts/plans.md +++ b/docs/concepts/plans.md @@ -458,6 +458,24 @@ These examples demonstrate how to select which models to restate based on model sqlmesh plan --restate-model "db.model_a" --start "2024-01-01" --end "2024-01-10" ``` +### Restatement after moving a model's start earlier + +When you move an incremental model's `start` earlier, SQLMesh backfills that model for the newly added date range, as usual. Downstream models may already hold intervals for that same range, though, and those were computed while the upstream model was empty there. + +To keep that stale data from lingering, SQLMesh also restates the downstream models' existing intervals for the added range. The range is widened to each downstream model's interval unit. The plan output lists these models under the heading "Models restated because the start of an upstream model moved earlier". + +No restatement happens when the new `start` is later than before or when the change is [breaking](#breaking-change), because downstream models are rebuilt anyway. + +SQLMesh skips a model and prints a single warning in these cases. Models downstream of a skipped model are not restated through it, since its data doesn't change, but they can still be restated through another upstream model: + +- The model has `disable_restatement` set to `true` (this is the default for SCD Type 2 models). +- The model's kind only supports restating its full history (`INCREMENTAL_BY_UNIQUE_KEY`, `INCREMENTAL_UNMANAGED`, `INCREMENTAL_BY_PARTITION`, and the SCD Type 2 kinds). The warning prints the `sqlmesh plan --restate-model ...` command to run manually. +- The plan targets a development environment and the downstream model's table is one that `prod` uses or will use. These models are restated when the change is applied to `prod`. + +If you use `--select-model`, `--skip-backfill` or `--empty-backfill`, stale models that the plan does not backfill are not restated either. The same applies to the part of the added range before the plan's `--start`, when it is set later than the new model `start` outside a development environment. In both cases the warning prints the `--restate-model` command to run manually. + +Unlike a plan with `--restate-model`, these automatic restatements do not make the plan a restatement plan. For example, [signals](../guides/signals.md) are evaluated as in a regular run. + ### Restating production vs development Restatement plans behave differently depending on if you're targeting the `prod` environment or a [development environment](./environments.md#how-to-use-environments). diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md index 47dcaaced1..89e7e6faf9 100644 --- a/docs/reference/model_configuration.md +++ b/docs/reference/model_configuration.md @@ -20,7 +20,7 @@ Configuration options for SQLMesh model properties. Supported by all model kinds | `tags` | Arbitrary strings used to organize or classify a model | array[str] | N | | `cron` | The cron expression specifying how often the model should be refreshed. (Default: `@daily`) | str | N | | `interval_unit` | The temporal granularity of the model's data intervals. Supported values: `year`, `month`, `day`, `hour`, `half_hour`, `quarter_hour`, `five_minute`. (Default: inferred from `cron`) | str | N | -| `start` | The date/time that determines the earliest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`. (Default: `yesterday`) | str \| int | N | +| `start` | The date/time that determines the earliest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`. (Default: `yesterday`). Moving `start` earlier also restates the existing intervals of downstream models for the added range; see [restatement after moving a model's start earlier](../concepts/plans.md#restatement-after-moving-a-models-start-earlier). | str \| int | N | | `end` | The date/time that determines the latest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`. | str \| int | N | | `description` | Description of the model. Automatically registered in the SQL engine's table COMMENT field or equivalent (if supported by the engine). | str | N | | `column_descriptions` | A key-value mapping of column names to column comments that will be registered in the SQL engine's table COMMENT field (if supported by the engine). Specified as key-value pairs (`column_name = 'column comment'`). If present, [inline column comments](../concepts/models/overview.md#inline-column-comments) will not be registered in the SQL engine. | dict | N | diff --git a/sqlmesh/core/analytics/collector.py b/sqlmesh/core/analytics/collector.py index cfdb60aadd..c13c96f659 100644 --- a/sqlmesh/core/analytics/collector.py +++ b/sqlmesh/core/analytics/collector.py @@ -171,7 +171,7 @@ def on_plan_apply_start( "no_gaps": plan.no_gaps, "forward_only": plan.forward_only, "ensure_finalized_snapshots": plan.ensure_finalized_snapshots, - "has_restatements": bool(plan.restatements), + "has_restatements": plan.is_restatement, "directly_modified_count": len(plan.directly_modified_snapshots), "indirectly_modified_count": len( { diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py index 8ae3be9083..36e3d32644 100644 --- a/sqlmesh/core/console.py +++ b/sqlmesh/core/console.py @@ -2064,12 +2064,31 @@ def _prompt_categorize( # We are computing dev previews, do not confuse the user by printing out something to do # with restatements. Dev previews are already highlighted in the backfill step pass - else: + + if not plan.restatements or ( + plan.start_extension_restatements and not plan.selected_models_to_restate + ): self.show_environment_difference_summary( plan.context_diff, no_diff=no_diff, ) + if plan.start_extension_restatements: + model_fqn_to_snapshot = {s.name: s for s in plan.snapshots.values()} + tree = Tree( + "[bold]Models restated because the start of an upstream model moved earlier:[/bold]\n" + "Their existing intervals for the added date range were computed without the new upstream data" + ) + for s_id in sorted(plan.start_extension_restatements): + tree.add( + model_fqn_to_snapshot[s_id.name].display_name( + plan.environment_naming_info, + default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, + dialect=self.dialect, + ) + ) + self._print(tree) + if plan.context_diff.has_changes: self.show_model_difference_summary( plan.context_diff, diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index a6307a9ffd..af281a189d 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -8,6 +8,7 @@ from datetime import datetime +from sqlmesh.core import constants as c from sqlmesh.core.console import PlanBuilderConsole, get_console from sqlmesh.core.config import ( AutoCategorizationMode, @@ -35,7 +36,7 @@ SnapshotChangeCategory, ) from sqlmesh.core.snapshot.categorizer import categorize_change -from sqlmesh.core.snapshot.definition import Interval, SnapshotId +from sqlmesh.core.snapshot.definition import Interval, SnapshotId, start_date from sqlmesh.utils import columns_to_types_all_known, random_id from sqlmesh.utils.dag import DAG from sqlmesh.utils.date import ( @@ -43,6 +44,8 @@ now, to_datetime, yesterday_ds, + to_ds, + to_ts, to_timestamp, time_like_to_str, is_relative, @@ -51,6 +54,8 @@ logger = logging.getLogger(__name__) +MS_PER_DAY = 24 * 60 * 60 * 1000 + class PlanBuilder: """Plan Builder constructs a Plan based on user choices for how they want to backfill, preview, etc. their changes. @@ -324,9 +329,10 @@ def build(self) -> Plan: else DeployabilityIndex.all_deployable() ) - restatements = self._build_restatements( + restatements, start_extension_restatements = self._build_restatements( dag, earliest_interval_start(self._context_diff.snapshots.values(), self.execution_time), + deployability_index, ) models_to_backfill = self._build_models_to_backfill(dag, restatements) @@ -363,7 +369,9 @@ def build(self) -> Plan: deployability_index=deployability_index, selected_models_to_restate=self._restate_models, restatements=restatements, - restate_all_snapshots=self._restate_all_snapshots, + start_extension_restatements=start_extension_restatements, + restate_all_snapshots=self._restate_all_snapshots + or self._restates_start_extension_in_prod(start_extension_restatements), start_override_per_model=self._start_override_per_model, end_override_per_model=end_override_per_model, selected_models_to_backfill=self._backfill_models, @@ -386,15 +394,27 @@ def _build_dag(self) -> DAG[SnapshotId]: return dag def _build_restatements( - self, dag: DAG[SnapshotId], earliest_interval_start: TimeLike - ) -> t.Dict[SnapshotId, Interval]: + self, + dag: DAG[SnapshotId], + earliest_interval_start: TimeLike, + deployability_index: DeployabilityIndex, + ) -> t.Tuple[t.Dict[SnapshotId, Interval], t.Set[SnapshotId]]: + """Builds the intervals to restate. + + Besides the models that were requested by a user (and their downstream models), this also covers + downstream models whose existing intervals are stale because the start of an upstream model moved earlier. + + Returns: + The restatements and the subset of the restated snapshots that are only restated because + of an upstream start extension. + """ restate_models = self._restate_models if restate_models == set(): # This is a warning but we print this as error since the Console is lacking API for warnings. self._console.log_error( "Provided restated models do not match any models. No models will be included in plan." ) - return {} + return {}, set() restatements: t.Dict[SnapshotId, Interval] = {} forward_only_preview_needed = self._forward_only_preview_needed @@ -417,8 +437,12 @@ def _build_restatements( } is_preview = True - if not restate_models: - return {} + start_extensions, cut_extensions = self._find_start_extensions() + if cut_extensions and not self._is_dev: + self._warn_about_cut_start_extensions(dag, cut_extensions, deployability_index) + if not restate_models and not start_extensions: + return {}, set() + restate_models = restate_models or set() start = self._start or earliest_interval_start end = self._end or now() @@ -428,24 +452,82 @@ def _build_restatements( if model_fqn not in self._model_fqn_to_snapshot: raise PlanError(f"Cannot restate model '{model_fqn}'. Model does not exist.") + backfill_scope = ( + self._build_models_to_backfill(dag, ()) if self._backfill_models is not None else None + ) + explicit_restatements: t.Set[SnapshotId] = set() + start_extension_restatements: t.Set[SnapshotId] = set() + # The range of an upstream start extension that reached each snapshot (restated or not). + start_extension_flow: t.Dict[SnapshotId, Interval] = {} + # Snapshots with stale intervals that are not restated, grouped by the reason. + outside_backfill_scope: t.List[Snapshot] = [] + shared_with_prod: t.List[Snapshot] = [] + full_history_only: t.List[Snapshot] = [] + stale_ranges: t.Dict[SnapshotId, Interval] = {} + start_cache: t.Dict[str, datetime] = {} + # Get restatement intervals for all restated snapshots and make sure that if an incremental snapshot expands it's # restatement range that it's downstream dependencies all expand their restatement ranges as well. for s_id in dag: snapshot = self._context_diff.snapshots[s_id] - if is_preview and snapshot.is_no_preview: - continue - # Since we are traversing the graph in topological order and the largest interval range is pushed down # the graph we just have to check our immediate parents in the graph and not the whole upstream graph. restating_parents = [ self._context_diff.snapshots[s] for s in snapshot.parents if s in restatements ] + extension_ranges = [ + r + for p in snapshot.parents + for r in (start_extensions.get(p), start_extension_flow.get(p)) + if r is not None + ] - if not restating_parents and snapshot.name not in restate_models: + is_explicit = not (is_preview and snapshot.is_no_preview) and ( + snapshot.name in restate_models + or any(p.snapshot_id in explicit_restatements for p in restating_parents) + ) + if not is_explicit and not extension_ranges: continue - if not forward_only_preview_needed: + extension_start = min((r[0] for r in extension_ranges), default=0) + extension_end = max((r[1] for r in extension_ranges), default=0) + if extension_ranges and ( + is_explicit + or to_timestamp( + start_date( + snapshot, + self._context_diff.snapshots, + start_cache, + relative_to=self.execution_time, + ) + ) + < extension_end + ): + # The extended range flows to the downstream models even if this snapshot isn't restated + # itself, as long as it holds (or will hold) data for that range. + start_extension_flow[s_id] = _align_to_interval_unit( + snapshot, extension_start, extension_end + ) + + if not is_explicit: + if self._is_dev and ( + not snapshot.is_paused or deployability_index.is_deployable(snapshot) + ): + # The physical table of this version is the one used in production. + if _has_intervals_in_range( + snapshot, extension_start, extension_end, deployability_index + ): + shared_with_prod.append(snapshot) + start_extension_flow.pop(s_id, None) + continue + # Only existing intervals can be stale. Anything else is backfilled as usual. + if not _has_intervals_in_range( + snapshot, extension_start, extension_end, deployability_index + ): + continue + + if not (forward_only_preview_needed and is_explicit): if self._is_dev and not snapshot.is_paused: self._console.log_warning( f"Cannot restate model '{snapshot.name}' because the current version is used in production. " @@ -457,44 +539,204 @@ def _build_restatements( f"Cannot restate model '{snapshot.name}'. " "Restatement is disabled for this model to prevent possible data loss. " "If you want to restate this model, change the model's `disable_restatement` setting to `false`." + + ( + "" + if is_explicit + else " Models downstream of it are not restated through it, since its data doesn't change." + ) ) + if not is_explicit: + start_extension_flow.pop(s_id, None) continue elif snapshot.is_seed: logger.info("Skipping restatement for model '%s'", snapshot.name) continue + if not is_explicit: + if snapshot.full_history_restatement_only and snapshot.is_incremental: + # Restating these models rebuilds their whole history, which is too expensive + # to do just because the start of an upstream model moved by a few intervals. + full_history_only.append(snapshot) + stale_ranges[s_id] = start_extension_flow.pop(s_id) + continue + if ( + self._skip_backfill + or self._empty_backfill + or (backfill_scope is not None and snapshot.name not in backfill_scope) + ): + outside_backfill_scope.append(snapshot) + stale_ranges[s_id] = start_extension_flow[s_id] + continue + possible_intervals = { restatements[p.snapshot_id] for p in restating_parents if p.is_incremental } - removal_start = ( - self._forward_only_preview_start(snapshot, start, end) if is_preview else start - ) - possible_intervals.add( - snapshot.get_removal_interval( - removal_start, - end, - self._execution_time, - strict=False, - is_preview=is_preview, + if is_explicit: + removal_start = ( + self._forward_only_preview_start(snapshot, start, end) if is_preview else start + ) + possible_intervals.add( + snapshot.get_removal_interval( + removal_start, + end, + self._execution_time, + strict=False, + is_preview=is_preview, + ) + ) + if extension_ranges: + possible_intervals.add( + snapshot.get_removal_interval( + extension_start, extension_end, self._execution_time, strict=False + ) ) - ) - snapshot_start = min(i[0] for i in possible_intervals) - snapshot_end = max(i[1] for i in possible_intervals) # We may be tasked with restating a time range smaller than the target snapshot interval unit # For example, restating an hour of Hourly Model A, which has a downstream dependency of Daily Model B # we need to ensure the whole affected day in Model B is restated - floored_snapshot_start = snapshot.node.interval_unit.cron_floor(snapshot_start) - floored_snapshot_end = snapshot.node.interval_unit.cron_floor(snapshot_end) - if to_timestamp(floored_snapshot_end) < snapshot_end: - snapshot_start = to_timestamp(floored_snapshot_start) - snapshot_end = to_timestamp( - snapshot.node.interval_unit.cron_next(floored_snapshot_end) - ) + restatements[s_id] = _align_to_interval_unit( + snapshot, + min(i[0] for i in possible_intervals), + max(i[1] for i in possible_intervals), + ) + if is_explicit: + explicit_restatements.add(s_id) + else: + start_extension_restatements.add(s_id) - restatements[s_id] = (snapshot_start, snapshot_end) + self._warn_about_stale_start_extensions( + outside_backfill_scope, + shared_with_prod, + full_history_only, + stale_ranges, + ) + + return restatements, start_extension_restatements + + def _find_start_extensions( + self, + ) -> t.Tuple[t.Dict[SnapshotId, Interval], t.Dict[SnapshotId, Interval]]: + """Finds the incremental models whose start moved earlier in this plan. + + Returns: + A mapping from the new snapshot ID to the [new start, old start) range that was added to its history, + and a mapping to the part of that range that was cut off because it's before the plan start. + Downstream models that already have intervals within this range computed them from incomplete data. + Whether those intervals are still around after the plan is applied (as opposed to being + rebuilt because of a breaking change) is decided when walking the downstream models. + """ + modified = self._context_diff.modified_snapshots + snapshots = self._context_diff.snapshots + # Unmodified snapshots are the same before and after the plan; this lets inferred starts + # be resolved through the parents that were in place before the plan. + previous_snapshots = {s_id: s for s_id, s in snapshots.items() if s.name not in modified} + previous_snapshots.update({old.snapshot_id: old for _, old in modified.values()}) + new_cache: t.Dict[str, datetime] = {} + old_cache: t.Dict[str, datetime] = {} + # Skip changes that don't move the start by a whole interval, e.g. an adjustment of the time of day. + plan_start = to_timestamp(self.start) if self.start else None + + extensions: t.Dict[SnapshotId, Interval] = {} + cut: t.Dict[SnapshotId, Interval] = {} + for new, old in modified.values(): + if not (new.is_model and old.is_model and new.is_incremental): + continue + if not _is_start_resolvable(old, previous_snapshots): + # The previous start would be guessed wrongly if some of its parents are unknown. + continue + interval_unit = new.node.interval_unit + new_start = start_date(new, snapshots, new_cache, relative_to=self.execution_time) + old_start = start_date( + old, previous_snapshots, old_cache, relative_to=self.execution_time + ) + start_ts = to_timestamp(interval_unit.cron_floor(new_start)) + end_ts = to_timestamp(interval_unit.cron_floor(old_start)) + if plan_start is not None and start_ts < min(plan_start, end_ts): + cut[new.snapshot_id] = (start_ts, min(plan_start, end_ts)) + if plan_start is not None: + start_ts = max(start_ts, plan_start) + if start_ts < end_ts: + extensions[new.snapshot_id] = (start_ts, end_ts) + return extensions, cut + + def _restates_start_extension_in_prod(self, snapshot_ids: t.Set[SnapshotId]) -> bool: + """Whether intervals of other versions of the snapshots also need to be cleared from state. + + This mirrors what is done for user requested production restatements. + """ + return not self._is_dev and any( + self._context_diff.snapshots[s_id].virtual_environment_mode.is_full + for s_id in snapshot_ids + ) - return restatements + def _warn_about_cut_start_extensions( + self, + dag: DAG[SnapshotId], + cut_extensions: t.Dict[SnapshotId, Interval], + deployability_index: DeployabilityIndex, + ) -> None: + """Warns about the part of an extension before the plan start, which is never detected again.""" + affected: t.Dict[SnapshotId, Snapshot] = {} + for s_id, (start, end) in cut_extensions.items(): + for downstream_id in dag.downstream(s_id): + downstream = self._context_diff.snapshots[downstream_id] + if _has_intervals_in_range(downstream, start, end, deployability_index): + affected[downstream_id] = downstream + if not affected: + return + start = min(r[0] for r in cut_extensions.values()) + end = max(r[1] for r in cut_extensions.values()) + restate_args = " ".join(f"--restate-model '{s.name}'" for s in affected.values()) + self._console.log_warning( + "The start of an upstream model moved earlier, but the plan start is after the beginning " + "of the added range, so downstream models with intervals in the cut off part were not " + f"restated: {', '.join(s.name for s in affected.values())}. " + f"To refresh them, run: sqlmesh plan {restate_args} {_range_args(start, end)}" + ) + + def _warn_about_stale_start_extensions( + self, + outside_backfill_scope: t.List[Snapshot], + shared_with_prod: t.List[Snapshot], + full_history_only: t.List[Snapshot], + flow: t.Dict[SnapshotId, Interval], + ) -> None: + intro = ( + "The start of an upstream model moved earlier, so some existing intervals of these " + "downstream models were computed without the added data" + ) + if shared_with_prod: + names = ", ".join(s.name for s in shared_with_prod) + self._console.log_warning( + f"{intro}: {names}. They use tables that production uses or will use, so this development " + "plan doesn't restate them. They are restated, along with their downstream models, when this change is applied to production." + ) + for snapshots, reason in ( + ( + outside_backfill_scope, + "they were not restated since the plan doesn't backfill them", + ), + ( + full_history_only, + "they only support restating their full history, so they were not restated automatically " + "(models downstream of them are not restated through them)", + ), + ): + if not snapshots: + continue + start = min(flow[s.snapshot_id][0] for s in snapshots) + end = max(flow[s.snapshot_id][1] for s in snapshots) + range_args = _range_args(start, end) + restate_args = " ".join(f"--restate-model '{s.name}'" for s in snapshots) + environment = ( + f" {self._context_diff.environment}" + if self._context_diff.environment != c.PROD + else "" + ) + self._console.log_warning( + f"{intro}, but {reason}: {', '.join(s.name for s in snapshots)}. " + f"To refresh them, run: sqlmesh plan{environment} {restate_args} {range_args}" + ) def _forward_only_preview_start( self, snapshot: Snapshot, default_start: TimeLike, end: TimeLike @@ -959,3 +1201,42 @@ def _modified_and_added_snapshots(self) -> t.List[Snapshot]: if snapshot.name in self._context_diff.modified_snapshots or snapshot.snapshot_id in self._context_diff.added ] + + +def _range_args(start: int, end: int) -> str: + """Formats the [start, end) range as command line arguments, where dates are inclusive.""" + if start % MS_PER_DAY == 0 and end % MS_PER_DAY == 0: + return f"--start {to_ds(start)} --end {to_ds(end - 1)}" + return f"--start '{to_ts(start)}' --end '{to_ts(end)}'" + + +def _has_intervals_in_range( + snapshot: Snapshot, start: int, end: int, deployability_index: DeployabilityIndex +) -> bool: + intervals = ( + snapshot.intervals + if deployability_index.is_deployable(snapshot) + else snapshot.dev_intervals + ) + return any(i_start < end and start < i_end for i_start, i_end in intervals) + + +def _is_start_resolvable(snapshot: Snapshot, snapshots: t.Dict[SnapshotId, Snapshot]) -> bool: + """Whether the effective start can be derived from the given snapshots (it may be inferred from parents).""" + if snapshot.node.start or not snapshot.parents: + return True + return all( + p in snapshots and _is_start_resolvable(snapshots[p], snapshots) for p in snapshot.parents + ) + + +def _align_to_interval_unit(snapshot: Snapshot, start: int, end: int) -> Interval: + """Widens the range if it ends in the middle of one of the snapshot's intervals.""" + interval_unit = snapshot.node.interval_unit + floored_end = interval_unit.cron_floor(end) + if to_timestamp(floored_end) < end: + return ( + to_timestamp(interval_unit.cron_floor(start)), + to_timestamp(interval_unit.cron_next(floored_end)), + ) + return start, end diff --git a/sqlmesh/core/plan/definition.py b/sqlmesh/core/plan/definition.py index 866299eff8..8d01a27950 100644 --- a/sqlmesh/core/plan/definition.py +++ b/sqlmesh/core/plan/definition.py @@ -67,6 +67,11 @@ class Plan(PydanticModel, frozen=True): Note that dev previews are also considered restatements, so :selected_models_to_restate can be empty while :restatements is still populated with dev previews """ + start_extension_restatements: t.Set[SnapshotId] = set() + """ + Models that are part of :restatements only because the start of one of their upstream models moved + earlier, i.e. the restatement wasn't requested by a user. + """ restate_all_snapshots: bool """Whether or not to clear intervals from state for other versions of the models listed in :restatements""" @@ -272,6 +277,7 @@ def to_evaluatable(self) -> EvaluatablePlan: skip_backfill=self.skip_backfill, empty_backfill=self.empty_backfill, restatements={s.name: i for s, i in self.restatements.items()}, + start_extension_restatements={s_id.name for s_id in self.start_extension_restatements}, restate_all_snapshots=self.restate_all_snapshots, is_dev=self.is_dev, allow_destructive_models=self.allow_destructive_models, @@ -317,6 +323,8 @@ class EvaluatablePlan(PydanticModel): skip_backfill: bool empty_backfill: bool restatements: t.Dict[str, Interval] + start_extension_restatements: t.Set[str] = set() + """Names of the models in :restatements that are only there because an upstream start moved earlier.""" restate_all_snapshots: bool is_dev: bool allow_destructive_models: t.Set[str] @@ -339,6 +347,11 @@ class EvaluatablePlan(PydanticModel): user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None selected_models: t.Optional[t.Set[str]] = None + @property + def is_restatement(self) -> bool: + """Whether the plan restates data on request (or previews), as opposed to just refreshing extended ranges.""" + return bool(set(self.restatements) - self.start_extension_restatements) + 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 71e5285d63..cc3fc8529e 100644 --- a/sqlmesh/core/plan/evaluator.py +++ b/sqlmesh/core/plan/evaluator.py @@ -261,7 +261,7 @@ def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePla allow_additive_snapshots=plan.allow_additive_models, selected_snapshot_ids=stage.selected_snapshot_ids, selected_models=plan.selected_models, - is_restatement=bool(plan.restatements), + is_restatement=plan.is_restatement, ) if errors: raise PlanError("Plan application failed.") diff --git a/sqlmesh/integrations/github/cicd/controller.py b/sqlmesh/integrations/github/cicd/controller.py index 58b0664c2d..9f9e288c97 100644 --- a/sqlmesh/integrations/github/cicd/controller.py +++ b/sqlmesh/integrations/github/cicd/controller.py @@ -528,7 +528,9 @@ def get_plan_summary(self, plan: Plan) -> str: try: # Clear out any output that might exist from prior steps self._console.consume_captured_output() - if plan.restatements: + if plan.restatements and not ( + plan.start_extension_restatements and not plan.selected_models_to_restate + ): self._console._print("\n**Restating models**\n") else: self._console.show_environment_difference_summary( diff --git a/tests/core/analytics/test_collector.py b/tests/core/analytics/test_collector.py index e87704e694..a574357dec 100644 --- a/tests/core/analytics/test_collector.py +++ b/tests/core/analytics/test_collector.py @@ -375,3 +375,25 @@ def test_on_migration(collector: AnalyticsCollector, mocker: MockerFixture): ), ] ) + + +def test_on_plan_apply_start_ignores_start_extension_restatements( + collector: AnalyticsCollector, init_and_plan_context: t.Callable +): + _, plan = init_and_plan_context("examples/sushi") + evaluatable = plan.to_evaluatable() + + def has_restatements(restatements: t.Dict, extensions: t.Set[str]) -> str: + updated = evaluatable.copy( + update={"restatements": restatements, "start_extension_restatements": extensions} + ) + collector._dispatcher.add_event.reset_mock() # type: ignore + collector.on_plan_apply_start( + plan=updated, engine_type="duckdb", state_sync_type="duckdb", scheduler_type="builtin" + ) + collector.flush() + event = collector._dispatcher.add_event.call_args.args[0]["event"] # type: ignore + return json.loads(event)["has_restatements"] + + assert has_restatements({"a": (0, 1)}, set()) is True + assert has_restatements({"a": (0, 1)}, {"a"}) is False diff --git a/tests/core/integration/test_restatement.py b/tests/core/integration/test_restatement.py index 245aa9d670..89754a02b1 100644 --- a/tests/core/integration/test_restatement.py +++ b/tests/core/integration/test_restatement.py @@ -39,7 +39,8 @@ from sqlmesh.utils.errors import ( ConflictingPlanError, ) -from tests.core.integration.utils import add_projection_to_model +from sqlmesh.core.snapshot import SnapshotChangeCategory +from tests.core.integration.utils import add_projection_to_model, plan_choice pytestmark = pytest.mark.slow @@ -1970,3 +1971,341 @@ def test_restatement_plan_outside_parent_date_range(init_and_plan_context: t.Cal ] context.apply(plan) + + +# --- Start extension (#4837): moving a model's `start` earlier must refresh downstream intervals --- + + +_EXTENSION_NOW = "2023-01-04 12:00:00 UTC" + + +def _extension_model( + name: str, + source: str, + start: str, + *, + kind: str = "INCREMENTAL_BY_TIME_RANGE (time_column ds)", + cron: str = "@daily", + extra_columns: str = "", + query: t.Optional[str] = None, +) -> str: + query = query or f"SELECT ds::DATE AS ds{extra_columns} FROM {source}" + if "FULL" not in kind: + query += " WHERE ds BETWEEN @start_ds AND @end_ds" + return f"MODEL (name s.{name}, kind {kind}, start '{start}', cron '{cron}');\n{query}" + + +class _ExtensionProject: + """A scratch project of models reading from `s.src` that can be rewritten between plans.""" + + def __init__(self, tmp_path: Path, models: t.Dict[str, str]) -> None: + self.models_dir = tmp_path / "models" + self.models_dir.mkdir() + self.write(models) + config = Config( + model_defaults=ModelDefaultsConfig(dialect="duckdb"), + default_connection=DuckDBConnectionConfig(), + ) + self.context = Context(paths=[tmp_path], config=config) + adapter = self.context.engine_adapter + adapter.create_schema("s") + adapter.execute( + "CREATE TABLE s.src AS SELECT range::DATE AS ds FROM " + "range(DATE '2022-12-25', DATE '2023-01-06', INTERVAL 1 DAY)" + ) + + def write(self, models: t.Dict[str, str]) -> None: + for name, definition in models.items(): + (self.models_dir / f"{name}.sql").write_text(definition) + + def update(self, models: t.Dict[str, str]) -> None: + self.write(models) + self.context.load() + + def ds(self, table: str) -> t.List[str]: + rows = self.context.engine_adapter.fetchall(f"SELECT ds FROM {table} ORDER BY ds") + return [str(r[0]) for r in rows] + + +@pytest.fixture +def capture_console() -> t.Iterator[CaptureTerminalConsole]: + original = get_console() + console = CaptureTerminalConsole() + set_console(console) + yield console + set_console(original) + + +def _chain(a_start: str, b_start: str = "2020-01-01", c_start: str = "2023-01-02", **b_kwargs): + return { + "a": _extension_model("a", "s.src", a_start), + "b": _extension_model("b", "s.a", b_start, **b_kwargs), + "c": _extension_model("c", "s.b", c_start), + } + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +@pytest.mark.parametrize( + "c_start_before, c_start_after", [("2023-01-02", "2023-01-01"), ("2020-01-01", "2020-01-01")] +) +def test_start_extension_refreshes_downstream_intervals( + tmp_path: Path, c_start_before: str, c_start_after: str +): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start=c_start_before)) + project.context.plan(auto_apply=True, no_prompts=True) + for table in ("a", "b", "c"): + assert project.ds(f"s.{table}") == ["2023-01-02", "2023-01-03"] + + project.update(_chain("2023-01-01", c_start=c_start_after)) + plan = project.context.plan(auto_apply=True, no_prompts=True) + + # b (and c, if its history already covers the range) are restated for the added range only. + restated = {s_id.name: interval for s_id, interval in plan.restatements.items()} + day = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")) + assert restated['"memory"."s"."b"'] == day + for table in ("a", "b", "c"): + assert project.ds(f"s.{table}") == ["2023-01-01", "2023-01-02", "2023-01-03"], table + + # Nothing is left to do. + assert not project.context.plan_builder().build().requires_backfill + + +@time_machine.travel("2023-01-02 12:00:00 UTC", tick=False) +def test_start_extension_restates_whole_coarser_downstream_interval(tmp_path: Path): + hourly_a = ( + "MODEL (name s.a, kind INCREMENTAL_BY_TIME_RANGE (time_column ts), start '{start}', " + "cron '@hourly');\nSELECT ts FROM s.hours WHERE ts BETWEEN @start_ts AND @end_ts" + ) + daily_b = ( + "MODEL (name s.b, kind INCREMENTAL_BY_TIME_RANGE (time_column ds), start '2023-01-01', " + "cron '@daily');\nSELECT ts::DATE AS ds, count(*) AS n FROM s.a " + "WHERE ts BETWEEN @start_ts AND @end_ts GROUP BY 1" + ) + project = _ExtensionProject( + tmp_path, {"a": hourly_a.format(start="2023-01-01 06:00:00"), "b": daily_b} + ) + project.context.engine_adapter.execute( + "CREATE TABLE s.hours AS SELECT range AS ts FROM " + "range(TIMESTAMP '2023-01-01 00:00:00', TIMESTAMP '2023-01-02 12:00:00', INTERVAL 1 HOUR)" + ) + project.context.plan(auto_apply=True, no_prompts=True) + count = "SELECT n FROM s.b WHERE ds = '2023-01-01'" + assert project.context.engine_adapter.fetchone(count) == (18,) + + project.update({"a": hourly_a.format(start="2023-01-01 00:00:00")}) + plan = project.context.plan(auto_apply=True, no_prompts=True) + + assert list(plan.restatements.values()) == [ + (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")) + ] + assert project.context.engine_adapter.fetchone(count) == (24,) + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_rebuilds_full_downstream_model(tmp_path: Path): + models = { + "a": _extension_model("a", "s.src", "2023-01-02"), + "b": _extension_model("b", "s.a", "2023-01-01", kind="FULL"), + } + project = _ExtensionProject(tmp_path, models) + project.context.plan(auto_apply=True, no_prompts=True) + assert project.ds("s.b") == ["2023-01-02", "2023-01-03"] + + project.update({"a": _extension_model("a", "s.src", "2023-01-01")}) + plan = project.context.plan(auto_apply=True, no_prompts=True) + + assert len(plan.restatements) == 1 + assert project.ds("s.b") == ["2023-01-01", "2023-01-02", "2023-01-03"] + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_skips_downstream_with_disabled_restatement( + tmp_path: Path, capture_console: CaptureTerminalConsole +): + kind = "INCREMENTAL_BY_TIME_RANGE (time_column ds, disable_restatement true)" + project = _ExtensionProject(tmp_path, _chain("2023-01-02", kind=kind)) + project.context.plan(auto_apply=True, no_prompts=True) + + project.update(_chain("2023-01-01", kind=kind)) + plan = project.context.plan(auto_apply=True, no_prompts=True) + + assert not plan.restatements + assert "Restatement is disabled for this model" in capture_console.captured_warnings + assert project.ds("s.a") == ["2023-01-01", "2023-01-02", "2023-01-03"] + assert project.ds("s.b") == ["2023-01-02", "2023-01-03"] + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_with_non_breaking_change_restates_extension_only(tmp_path: Path): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + + # Add a column to `a` (non-breaking) and move its start earlier. + project.update({"a": _extension_model("a", "s.src", "2023-01-01", extra_columns=", 1 AS one")}) + plan_builder = project.context.plan_builder(no_auto_categorization=True) + plan_choice(plan_builder, SnapshotChangeCategory.NON_BREAKING) + plan = plan_builder.build() + project.context.apply(plan) + + day = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")) + assert {s_id.name: i for s_id, i in plan.restatements.items()} == { + '"memory"."s"."b"': day, + '"memory"."s"."c"': day, + } + for table in ("a", "b", "c"): + assert project.ds(f"s.{table}") == ["2023-01-01", "2023-01-02", "2023-01-03"], table + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_with_breaking_change_needs_no_restatement(tmp_path: Path): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + + project.update({"a": _extension_model("a", "s.src", "2023-01-01", extra_columns=", 1 AS one")}) + plan_builder = project.context.plan_builder(no_auto_categorization=True) + plan_choice(plan_builder, SnapshotChangeCategory.BREAKING) + plan = plan_builder.build() + project.context.apply(plan) + + assert not plan.restatements + for table in ("a", "b", "c"): + assert project.ds(f"s.{table}") == ["2023-01-01", "2023-01-02", "2023-01-03"], table + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_moved_later_does_not_restate(tmp_path: Path): + project = _ExtensionProject(tmp_path, _chain("2023-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + + project.update(_chain("2023-01-02")) + plan = project.context.plan(auto_apply=True, no_prompts=True) + assert not plan.restatements + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_with_select_model_warns_about_stale_models( + tmp_path: Path, capture_console: CaptureTerminalConsole +): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + + project.update(_chain("2023-01-01", c_start="2020-01-01")) + plan = project.context.plan(select_models=["s.a"], auto_apply=True, no_prompts=True) + + assert not plan.restatements + warnings = capture_console.captured_warnings + assert '"memory"."s"."b"' in warnings and "--restate-model" in warnings + assert project.ds("s.b") == ["2023-01-02", "2023-01-03"] + + # Following the advice from the warning fixes the downstream data. + project.context.plan( + restate_models=["s.b"], + start="2023-01-01", + end="2023-01-01", + auto_apply=True, + no_prompts=True, + ) + assert project.ds("s.b") == ["2023-01-01", "2023-01-02", "2023-01-03"] + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_clears_intervals_of_derived_dev_versions(tmp_path: Path): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + context = project.context + context.plan(auto_apply=True, no_prompts=True) + + # dev gets its own version of b, which has the (soon to be stale) intervals of its own. + project.update(_chain("2023-01-02", c_start="2020-01-01", extra_columns=", 1 AS one")) + context.plan("dev", auto_apply=True, no_prompts=True) + dev_b = context.get_snapshot("s.b") + assert dev_b is not None + assert dev_b.intervals + + # Back to prod's version of b, with the extended start of a. + project.update(_chain("2023-01-01", c_start="2020-01-01")) + context.plan("prod", auto_apply=True, no_prompts=True) + + day = (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")) + state_b = context.state_sync.get_snapshots([dev_b.snapshot_id])[dev_b.snapshot_id] + assert not any(start <= day[0] and day[1] <= end for start, end in state_b.intervals) + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_is_explained_in_plan_output( + tmp_path: Path, capture_console: CaptureTerminalConsole +): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + capture_console.consume_captured_output() + + project.update(_chain("2023-01-01", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True, skip_tests=True) + output = capture_console.consume_captured_output() + + assert "Differences from the `prod` environment" in output + assert "restated because the start of an upstream model moved earlier" in output + assert "s.b" in output and "s.c" in output + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_in_dev_does_not_rewrite_prod_tables(tmp_path: Path): + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + context = project.context + context.plan(auto_apply=True, no_prompts=True) + + project.update(_chain("2023-01-01", c_start="2020-01-01")) + dev_plan = context.plan("dev", auto_apply=True, no_prompts=True, skip_tests=True) + + # b and c share their tables with prod, so the dev plan leaves them alone. + assert not dev_plan.restatements + assert project.ds("s.a") == ["2023-01-01", "2023-01-02", "2023-01-03"] + assert project.ds("s.b") == ["2023-01-02", "2023-01-03"] + assert project.ds("s.c") == ["2023-01-02", "2023-01-03"] + + # They are restated when the change is deployed to prod. + prod_plan = context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True) + assert {s_id.name for s_id in prod_plan.restatements} == { + '"memory"."s"."b"', + '"memory"."s"."c"', + } + assert project.ds("s.b") == ["2023-01-01", "2023-01-02", "2023-01-03"] + assert project.ds("s.c") == ["2023-01-01", "2023-01-02", "2023-01-03"] + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_flows_through_view(tmp_path: Path): + models = { + "a": _extension_model("a", "s.src", "2023-01-02"), + "b": "MODEL (name s.b, kind VIEW);\nSELECT ds FROM s.a", + "c": _extension_model("c", "s.b", "2020-01-01"), + } + project = _ExtensionProject(tmp_path, models) + project.context.plan(auto_apply=True, no_prompts=True) + assert project.ds("s.c") == ["2023-01-02", "2023-01-03"] + + project.update({"a": _extension_model("a", "s.src", "2023-01-01")}) + project.context.plan(auto_apply=True, no_prompts=True) + assert project.ds("s.c") == ["2023-01-01", "2023-01-02", "2023-01-03"] + + +@time_machine.travel(_EXTENSION_NOW, tick=False) +def test_start_extension_does_not_run_plan_as_restatement(tmp_path: Path, mocker): + from sqlmesh.core.scheduler import Scheduler + + spy = mocker.spy(Scheduler, "run_merged_intervals") + project = _ExtensionProject(tmp_path, _chain("2023-01-02", c_start="2020-01-01")) + project.context.plan(auto_apply=True, no_prompts=True) + + project.update(_chain("2023-01-01", c_start="2020-01-01")) + plan = project.context.plan(auto_apply=True, no_prompts=True) + assert plan.restatements + assert spy.call_args.kwargs["is_restatement"] is False + + project.context.plan( + restate_models=["s.b"], + start="2023-01-01", + end="2023-01-01", + auto_apply=True, + no_prompts=True, + ) + assert spy.call_args.kwargs["is_restatement"] is True diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py index add85d2eec..a5c9f4bdef 100644 --- a/tests/core/test_plan.py +++ b/tests/core/test_plan.py @@ -4510,3 +4510,582 @@ def test_forward_only_indirect_change_to_materialized_view(make_snapshot): # Forward-only indirect changes to MVs should not always be classified as indirect breaking. # Instead, we want to preserve the standard categorization. assert snapshot_b_new.change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING + + +# --- Start extension (#4837): extending a model's `start` earlier must refresh downstream intervals --- + +_EXECUTION_TIME = "2023-01-04 12:00:00 UTC" + + +def _ts(value: str) -> int: + return to_timestamp(value) + + +def _start_extension_context_diff( + make_snapshot: t.Callable, + specs: t.List[t.Dict[str, t.Any]], + environment: str = "prod", +) -> t.Tuple[ContextDiff, t.Dict[str, Snapshot], t.Dict[str, Snapshot]]: + """Builds a context diff in which every listed model has an old and a new version. + + Each spec is a dict with `name`, an optional `parent`, `old` / `new` model kwargs overrides and + optional `intervals` (a list of (start, end) dates added to the old snapshot). + """ + old_nodes: t.Dict[str, t.Any] = {} + new_nodes: t.Dict[str, t.Any] = {} + olds: t.Dict[str, Snapshot] = {} + news: t.Dict[str, Snapshot] = {} + + for spec in specs: + name = spec["name"] + parent = spec.get("parent") + if spec.get("parents"): + query = parse_one(" union all ".join(f"select ds from {p}" for p in spec["parents"])) + elif parent and spec.get("self_referencing"): + query = parse_one(f"select ds from {parent} union all select ds from {name}") + else: + query = parse_one(f"select ds from {parent}" if parent else "select 1 as ds") + common: t.Dict[str, t.Any] = { + "name": name, + "query": query, + "kind": spec.get("kind", IncrementalByTimeRangeKind(time_column="ds")), + "start": spec.get("start", "2020-01-01"), + "cron": spec.get("cron", "@daily"), + } + old = make_snapshot(SqlModel(**{**common, **spec.get("old", {})}), nodes=dict(old_nodes)) + new = make_snapshot(SqlModel(**{**common, **spec.get("new", {})}), nodes=dict(new_nodes)) + old.categorize_as(SnapshotChangeCategory.BREAKING) + new.previous_versions = old.all_versions + for start, end in spec.get("intervals", []): + old.add_interval(start, end) + if spec.get("unpaused"): + new.unpaused_ts = 1 + + old_nodes[old.name] = old.model + new_nodes[new.name] = new.model + olds[name] = old + news[name] = new + + context_diff = ContextDiff( + environment=environment, + 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={new.name: (new, olds[n]) for n, new in news.items()}, + snapshots={s.snapshot_id: s for s in news.values()}, + new_snapshots={s.snapshot_id: s for s in news.values()}, + previous_plan_id=None, + previously_promoted_snapshot_ids=set(), + previous_finalized_snapshots=None, + previous_gateway_managed_virtual_layer=False, + gateway_managed_virtual_layer=False, + environment_statements=[], + ) + return context_diff, news, olds + + +def _chain_specs( + a_old: str = "2023-01-02", + a_new: str = "2023-01-01", + **child_overrides: t.Any, +) -> t.List[t.Dict[str, t.Any]]: + """a (start moves) -> b -> c where b and c have long-standing intervals.""" + intervals = [("2020-01-01", "2023-01-03")] + return [ + { + "name": "a", + "old": {"start": a_old}, + "new": {"start": a_new}, + "intervals": [(a_old, "2023-01-03")], + }, + {"name": "b", "parent": "a", "intervals": intervals, **child_overrides}, + {"name": "c", "parent": "b", "intervals": intervals}, + ] + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_restates_downstream(make_snapshot): + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + plan = PlanBuilder(context_diff).build() + + expected = (_ts("2023-01-01"), _ts("2023-01-02")) + assert plan.restatements == { + snaps["b"].snapshot_id: expected, + snaps["c"].snapshot_id: expected, + } + # The model whose start moved is backfilled as usual and is not restated itself. + assert snaps["a"].snapshot_id not in plan.restatements + # This is not a user requested restatement. + assert not plan.selected_models_to_restate + assert plan.models_to_backfill is None + assert plan.requires_backfill + assert plan.start_extension_restatements == {snaps["b"].snapshot_id, snaps["c"].snapshot_id} + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_moved_later_does_not_restate(make_snapshot): + context_diff, _, _ = _start_extension_context_diff( + make_snapshot, _chain_specs(a_old="2023-01-01", a_new="2023-01-02") + ) + plan = PlanBuilder(context_diff).build() + assert not plan.restatements + assert not plan.start_extension_restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_without_downstream_is_a_noop(make_snapshot): + context_diff, _, _ = _start_extension_context_diff(make_snapshot, _chain_specs()[:1]) + plan = PlanBuilder(context_diff).build() + assert not plan.restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_downstream_starting_after_range_is_not_restated(make_snapshot): + specs = _chain_specs() + specs[1].update(start="2023-01-03", intervals=[("2023-01-03", "2023-01-03")]) + specs[2].update(start="2023-01-03", intervals=[("2023-01-03", "2023-01-03")]) + context_diff, _, _ = _start_extension_context_diff(make_snapshot, specs) + assert not PlanBuilder(context_diff).build().restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_unchanged_relative_start_is_not_an_extension(make_snapshot): + specs = _chain_specs() + specs[0].update( + old={"start": "2 days ago", "description": "old"}, + new={"start": "2 days ago", "description": "new"}, + ) + context_diff, _, _ = _start_extension_context_diff(make_snapshot, specs) + assert not PlanBuilder(context_diff).build().restatements + + # The same relative expression resolves against the plan's execution time, not the clock + # at the time the snapshots were created. + assert not PlanBuilder(context_diff, execution_time="2023-02-01").build().restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_relative_start_moved_earlier(make_snapshot): + specs = _chain_specs() + specs[0].update(old={"start": "2 days ago"}, new={"start": "3 days ago"}) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + + expected = (_ts("3 days ago"), _ts("2 days ago")) + assert plan.restatements[snaps["b"].snapshot_id] == expected + assert plan.restatements[snaps["c"].snapshot_id] == expected + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_widens_range_for_coarser_downstream(make_snapshot): + specs = _chain_specs(a_old="2023-01-01 06:00:00", a_new="2023-01-01 00:00:00") + specs[0]["cron"] = "@hourly" + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + + # The six hours that were added belong to a daily interval, which gets restated as a whole. + whole_day = (_ts("2023-01-01"), _ts("2023-01-02")) + assert plan.restatements[snaps["b"].snapshot_id] == whole_day + assert plan.restatements[snaps["c"].snapshot_id] == whole_day + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_full_downstream_is_restated(make_snapshot): + context_diff, snaps, _ = _start_extension_context_diff( + make_snapshot, _chain_specs(kind=FullKind()) + ) + plan = PlanBuilder(context_diff).build() + assert snaps["b"].snapshot_id in plan.restatements + # Descendants of the FULL model still only get the affected range. + assert plan.restatements[snaps["c"].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-02")) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_respects_disable_restatement(make_snapshot, mocker: MockerFixture): + context_diff, snaps, _ = _start_extension_context_diff( + make_snapshot, + _chain_specs(kind=IncrementalByTimeRangeKind(time_column="ds", disable_restatement=True)), + ) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console).build() + + # b keeps its data in this plan, so c (derived from it) is consistent with it. + assert not plan.restatements + console.log_warning.assert_called_once() + message = console.log_warning.call_args[0][0] + assert "Restatement is disabled for this model" in message + assert "Models downstream of it are not restated through it" in message + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_no_spurious_warning_without_existing_intervals( + make_snapshot, mocker: MockerFixture +): + # SCD Type 2 has `disable_restatement` enabled by default, but there is nothing stale to + # restate if the model has no intervals in the affected range, so there must be no warning. + specs = _chain_specs( + kind=dict(name=ModelKindName.SCD_TYPE_2_BY_TIME, unique_key=["ds"]), + start="2023-01-02", + intervals=[("2023-01-02", "2023-01-03")], + ) + context_diff, _, _ = _start_extension_context_diff(make_snapshot, specs) + console = mocker.Mock() + assert not PlanBuilder(context_diff, console=console).build().restatements + console.log_warning.assert_not_called() + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_scd_type_2_downstream_warns(make_snapshot, mocker: MockerFixture): + specs = _chain_specs(kind=dict(name=ModelKindName.SCD_TYPE_2_BY_TIME, unique_key=["ds"])) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + console = mocker.Mock() + assert ( + snaps["b"].snapshot_id + not in PlanBuilder(context_diff, console=console).build().restatements + ) + assert "Restatement is disabled" in console.log_warning.call_args[0][0] + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_full_history_only_downstream_warns_once( + make_snapshot, mocker: MockerFixture +): + specs = _chain_specs(kind=dict(name=ModelKindName.INCREMENTAL_BY_UNIQUE_KEY, unique_key=["ds"])) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console).build() + + # A one day start move must not rebuild the full history of b automatically. + assert snaps["b"].snapshot_id not in plan.restatements + console.log_warning.assert_called_once() + message = console.log_warning.call_args[0][0] + assert snaps["b"].name in message and "full history" in message + assert f"--restate-model '{snaps['b'].name}'" in message + assert "--start 2023-01-01 --end 2023-01-01" in message + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_dev_does_not_restate_versions_shared_with_prod( + make_snapshot, mocker: MockerFixture +): + # The new (paused) versions of b and c are metadata only changes: they share their physical + # tables with prod, so a dev plan must not rewrite those. + context_diff, snaps, _ = _start_extension_context_diff( + make_snapshot, _chain_specs(), environment="dev" + ) + console = mocker.Mock() + plan = PlanBuilder(context_diff, is_dev=True, console=console).build() + + assert not plan.restatements + assert not plan.start_extension_restatements + console.log_warning.assert_called_once() + message = console.log_warning.call_args[0][0] + assert snaps["b"].name in message + assert "tables that production uses or will use" in message + assert "along with their downstream models" in message + assert "Cannot restate model" not in message + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_dev_restates_versions_with_their_own_table( + make_snapshot, mocker: MockerFixture +): + # Forward-only changes are previewed in dev tables (initialized from the production intervals), + # so restating them can't affect prod. + specs = _chain_specs() + specs[0]["new"]["query"] = parse_one("select 1 as ds, 2 as extra") + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs, environment="dev") + console = mocker.Mock() + plan = PlanBuilder( + context_diff, is_dev=True, forward_only=True, start="2023-01-01", console=console + ).build() + + assert not plan.deployability_index.is_deployable(snaps["b"]) + assert {s_id.name for s_id in plan.start_extension_restatements} == { + snaps["b"].name, + snaps["c"].name, + } + assert snaps["b"].dev_intervals + console.log_warning.assert_not_called() + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_flows_through_view_without_intervals(make_snapshot): + specs = _chain_specs(kind=ViewKind(), intervals=[]) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + # c is derived from the view, which reflects the new data of a right away. + assert plan.restatements[snaps["c"].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-02")) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_flows_through_embedded_model(make_snapshot): + specs = _chain_specs(kind=dict(name=ModelKindName.EMBEDDED), intervals=[]) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + assert plan.restatements[snaps["c"].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-02")) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_flow_stops_at_models_that_do_not_cover_the_range(make_snapshot): + # b starts after the added range, so it has no data in it and neither does anything derived from it. + specs = _chain_specs( + kind=ViewKind(), start="2023-01-03", intervals=[("2023-01-03", "2023-01-03")] + ) + context_diff, _, _ = _start_extension_context_diff(make_snapshot, specs) + assert not PlanBuilder(context_diff).build().restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_flow_stops_at_child_skipped_by_a_guard( + make_snapshot, mocker: MockerFixture +): + specs = _chain_specs(kind=dict(name=ModelKindName.INCREMENTAL_BY_UNIQUE_KEY, unique_key=["ds"])) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console).build() + + # b is warn-only and unchanged by this plan, so c is consistent with it: no second warning either. + assert not plan.restatements + console.log_warning.assert_called_once() + assert ( + "models downstream of them are not restated through them" + in console.log_warning.call_args[0][0] + ) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_flows_through_child_that_is_backfilled_in_this_plan(make_snapshot): + # b has no intervals at all so it is backfilled for the added range, and c was computed from the old b. + specs = _chain_specs(intervals=[]) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + assert set(plan.restatements) == {snaps["c"].snapshot_id} + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_ignores_unresolvable_inferred_start(make_snapshot): + # b has no start and one of its old parents is gone, so its old start can't be known. + specs = _chain_specs() + specs[1].update(start=None) + specs[2].update(start=None) + context_diff, snaps, olds = _start_extension_context_diff(make_snapshot, specs) + # a is no longer reported as modified, so the old version of a that b was derived from can't be found. + context_diff.modified_snapshots = { + name: pair + for name, pair in context_diff.modified_snapshots.items() + if name != snaps["a"].name + } + assert olds["a"].snapshot_id not in context_diff.snapshots + + plan = PlanBuilder(context_diff).build() + assert not plan.restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_before_plan_start_is_ignored(make_snapshot): + context_diff, _, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + assert not PlanBuilder(context_diff, is_dev=True, start="2023-01-02").build().restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_within_the_same_interval_unit_is_ignored(make_snapshot): + context_diff, _, _ = _start_extension_context_diff( + make_snapshot, _chain_specs(a_old="2023-01-01 12:00:00", a_new="2023-01-01 00:00:00") + ) + assert not PlanBuilder(context_diff).build().restatements + + +def test_evaluatable_plan_is_restatement_ignores_start_extensions(make_snapshot): + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + with time_machine.travel(_EXECUTION_TIME): + plan = PlanBuilder(context_diff).build() + evaluatable = plan.to_evaluatable() + assert evaluatable.restatements + assert evaluatable.start_extension_restatements == {snaps["b"].name, snaps["c"].name} + assert not evaluatable.is_restatement + + explicit = PlanBuilder(context_diff, restate_models={snaps["a"].name}).build() + assert explicit.to_evaluatable().is_restatement + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_with_breaking_change_does_not_restate(make_snapshot): + specs = _chain_specs() + specs[0]["new"]["query"] = parse_one("select 1 as ds, 2 as extra") + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + builder = PlanBuilder(context_diff, auto_categorization_enabled=False) + builder.set_choice(snaps["a"], SnapshotChangeCategory.BREAKING) + plan = builder.build() + + # Downstream models get new versions and are rebuilt from scratch: no extra work is needed. + assert snaps["b"].change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert not plan.restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_with_non_breaking_change_restates_extension_only(make_snapshot): + specs = _chain_specs() + specs[0]["new"]["query"] = parse_one("select 1 as ds, 2 as extra") + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + builder = PlanBuilder(context_diff, auto_categorization_enabled=False) + builder.set_choice(snaps["a"], SnapshotChangeCategory.NON_BREAKING) + plan = builder.build() + + assert snaps["b"].change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING + expected = (_ts("2023-01-01"), _ts("2023-01-02")) + assert plan.restatements == { + snaps["b"].snapshot_id: expected, + snaps["c"].snapshot_id: expected, + } + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_inferred_start_is_resolved_through_parents(make_snapshot): + # b has no explicit start so its start moves together with a's. It has no intervals in the + # affected range itself, but c (explicit start) does and is derived from b. + specs = _chain_specs() + specs[1].update(start=None, intervals=[("2023-01-02", "2023-01-03")]) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + + assert set(plan.restatements) == {snaps["c"].snapshot_id} + assert plan.restatements[snaps["c"].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-02")) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_depends_on_past_downstream_restates_through_execution_time( + make_snapshot, +): + specs = _chain_specs() + specs[1]["self_referencing"] = True + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + + assert snaps["b"].depends_on_past + start, end = plan.restatements[snaps["b"].snapshot_id] + assert start == _ts("2023-01-01") + assert end >= _ts(_EXECUTION_TIME) + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_forward_only_downstream_follows_regular_rules(make_snapshot): + specs = _chain_specs(kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True)) + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + plan = PlanBuilder(context_diff).build() + assert snaps["b"].snapshot_id in plan.restatements + assert snaps["c"].snapshot_id in plan.restatements + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_merges_with_explicit_restatement(make_snapshot): + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + plan = PlanBuilder( + context_diff, restate_models={snaps["a"].name}, start="2023-01-01", end="2023-01-03" + ).build() + + # The extension range is covered by the explicit range: there is a single, merged range per model. + assert plan.restatements[snaps["a"].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-04")) + for name in ("b", "c"): + assert plan.restatements[snaps[name].snapshot_id] == (_ts("2023-01-01"), _ts("2023-01-04")) + # Only explicitly requested models are listed as such. + assert plan.selected_models_to_restate == {snaps["a"].name} + assert plan.start_extension_restatements == set() + + +@time_machine.travel(_EXECUTION_TIME) +@pytest.mark.parametrize("flag", ["select_model", "skip_backfill", "empty_backfill"]) +def test_start_extension_warns_instead_of_restating_when_backfill_is_limited( + make_snapshot, mocker: MockerFixture, flag: str +): + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + kwargs: t.Dict[str, t.Any] = ( + {"backfill_models": {snaps["a"].name}} if flag == "select_model" else {flag: True} + ) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console, **kwargs).build() + + assert not plan.restatements + message = console.log_warning.call_args[0][0] + assert snaps["b"].name in message and snaps["c"].name in message + assert "--restate-model" in message + assert "--start 2023-01-01 --end 2023-01-01" in message + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_with_select_model_restates_selected_downstream( + make_snapshot, mocker: MockerFixture +): + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, _chain_specs()) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console, backfill_models={snaps["b"].name}).build() + + # b is selected (c isn't), so only b is restated and the plan warns about c. + assert set(plan.restatements) == {snaps["b"].snapshot_id} + message = console.log_warning.call_args[0][0] + assert snaps["c"].name in message and snaps["b"].name not in message + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_in_diamond_flows_through_unguarded_branch( + make_snapshot, mocker: MockerFixture +): + # a -> b (restatement disabled) -> d and a -> c -> d: d is restated through c only. + intervals = [("2020-01-01", "2023-01-03")] + specs: t.List[t.Dict[str, t.Any]] = [ + { + "name": "a", + "old": {"start": "2023-01-02"}, + "new": {"start": "2023-01-01"}, + "intervals": [("2023-01-02", "2023-01-03")], + }, + { + "name": "b", + "parent": "a", + "kind": IncrementalByTimeRangeKind(time_column="ds", disable_restatement=True), + "intervals": intervals, + }, + {"name": "c", "parent": "a", "intervals": intervals}, + {"name": "d", "parents": ["b", "c"], "intervals": intervals}, + ] + context_diff, snaps, _ = _start_extension_context_diff(make_snapshot, specs) + console = mocker.Mock() + plan = PlanBuilder(context_diff, console=console).build() + + day = (_ts("2023-01-01"), _ts("2023-01-02")) + assert plan.restatements == {snaps["c"].snapshot_id: day, snaps["d"].snapshot_id: day} + console.log_warning.assert_called_once() + assert snaps["b"].name in console.log_warning.call_args[0][0] + + +@time_machine.travel(_EXECUTION_TIME) +def test_start_extension_cut_by_plan_start_warns_in_prod_only(make_snapshot, mocker: MockerFixture): + # The range [01-01, 01-03) is cut to [01-02, 01-03) by the plan start. The remainder won't be + # detected again once this plan is applied. + def build(is_dev: bool, console) -> Plan: + context_diff, snaps, _ = _start_extension_context_diff( + make_snapshot, + _chain_specs(a_old="2023-01-03", a_new="2023-01-01"), + environment="dev" if is_dev else "prod", + ) + return PlanBuilder( + context_diff, + is_dev=is_dev, + start="2023-01-02", + restate_models=None if is_dev else {snaps["a"].name}, + console=console, + ).build() + + console = mocker.Mock() + build(False, console) + console.log_warning.assert_called_once() + message = console.log_warning.call_args[0][0] + assert "--restate-model '\"b\"' --restate-model '\"c\"'" in message + assert "--start 2023-01-01 --end 2023-01-01" in message + + console = mocker.Mock() + build(True, console) + assert not any("cut off" in call[0][0] for call in console.log_warning.call_args_list) diff --git a/tests/integrations/github/cicd/test_github_controller.py b/tests/integrations/github/cicd/test_github_controller.py index dbb09f4336..432cde3273 100644 --- a/tests/integrations/github/cicd/test_github_controller.py +++ b/tests/integrations/github/cicd/test_github_controller.py @@ -961,3 +961,27 @@ def test_chunk_up_api_message_preserves_multibyte_chars( assert "".join(chunks) == message # Each chunk still respects the GitHub API byte limit. assert all(len(chunk.encode("utf-8")) <= max_bytes for chunk in chunks) + + +@time_machine.travel("2025-07-07 00:00:00 UTC", tick=False) +def test_get_plan_summary_for_start_extension_restatements_shows_differences( + github_client, make_controller: t.Callable[..., GithubController] +): + controller = make_controller( + "tests/fixtures/github/pull_request_synchronized.json", + github_client, + mock_out_context=False, + ) + plan = controller.prod_plan + s_id = next(iter(plan.snapshots)) + + explicit = plan.copy(update={"restatements": {s_id: (0, 1)}}) + assert "Restating models" in controller.get_plan_summary(explicit) + + # Restatements that only exist because a start moved earlier don't replace the diff summary. + extension = plan.copy( + update={"restatements": {s_id: (0, 1)}, "start_extension_restatements": {s_id}} + ) + summary = controller.get_plan_summary(extension) + assert "Restating models" not in summary + assert "**Added Models:**" in summary