sqlmesh.core.plan.builder

  1from __future__ import annotations
  2
  3import logging
  4import re
  5import typing as t
  6from collections import defaultdict
  7from functools import cached_property
  8from datetime import datetime
  9
 10
 11from sqlmesh.core.console import PlanBuilderConsole, get_console
 12from sqlmesh.core.config import (
 13    AutoCategorizationMode,
 14    CategorizerConfig,
 15    EnvironmentSuffixTarget,
 16)
 17from sqlmesh.core.context_diff import ContextDiff
 18from sqlmesh.core.environment import EnvironmentNamingInfo
 19from sqlmesh.core.plan.common import should_force_rebuild, is_breaking_kind_change
 20from sqlmesh.core.plan.definition import (
 21    Plan,
 22    SnapshotMapping,
 23    UserProvidedFlags,
 24    earliest_interval_start,
 25)
 26from sqlmesh.core.schema_diff import (
 27    get_schema_differ,
 28    has_drop_alteration,
 29    has_additive_alteration,
 30    TableAlterOperation,
 31)
 32from sqlmesh.core.snapshot import (
 33    DeployabilityIndex,
 34    Snapshot,
 35    SnapshotChangeCategory,
 36)
 37from sqlmesh.core.snapshot.categorizer import categorize_change
 38from sqlmesh.core.snapshot.definition import Interval, SnapshotId
 39from sqlmesh.utils import columns_to_types_all_known, random_id
 40from sqlmesh.utils.dag import DAG
 41from sqlmesh.utils.date import (
 42    TimeLike,
 43    now,
 44    to_datetime,
 45    yesterday_ds,
 46    to_timestamp,
 47    time_like_to_str,
 48    is_relative,
 49)
 50from sqlmesh.utils.errors import NoChangesPlanError, PlanError
 51
 52logger = logging.getLogger(__name__)
 53
 54
 55class PlanBuilder:
 56    """Plan Builder constructs a Plan based on user choices for how they want to backfill, preview, etc. their changes.
 57
 58    Args:
 59        context_diff: The context diff that the plan is based on.
 60        start: The start time to backfill data.
 61        end: The end time to backfill data.
 62        execution_time: The date/time time reference to use for execution time. Defaults to now.
 63            If :start or :end are relative time expressions, they are interpreted as relative to the :execution_time
 64        apply: The callback to apply the plan.
 65        restate_models: A list of models for which the data should be restated for the time range
 66            specified in this plan. Note: models defined outside SQLMesh (external) won't be a part
 67            of the restatement.
 68        restate_all_snapshots: If restatements are present, this flag indicates whether or not the intervals
 69            being restated should be cleared from state for other versions of this model (typically, versions that are present in other environments).
 70            If set to None, the default behaviour is to not clear anything unless the target environment is prod.
 71        backfill_models: A list of fully qualified model names for which the data should be backfilled as part of this plan.
 72        no_gaps:  Whether to ensure that new snapshots for nodes that are already a
 73            part of the target environment have no data gaps when compared against previous
 74            snapshots for same nodes.
 75        skip_backfill: Whether to skip the backfill step.
 76        empty_backfill: Like skip_backfill, but also records processed intervals.
 77        is_dev: Whether this plan is for development purposes.
 78        forward_only: Whether the purpose of the plan is to make forward only changes.
 79        allow_destructive_models: A list of fully qualified model names whose forward-only changes are allowed to be destructive.
 80        allow_additive_models: A list of fully qualified model names whose forward-only changes are allowed to be additive.
 81        environment_ttl: The period of time that a development environment should exist before being deleted.
 82        categorizer_config: Auto categorization settings.
 83        auto_categorization_enabled: Whether to apply auto categorization.
 84        effective_from: The effective date from which to apply forward-only changes on production.
 85        include_unmodified: Indicates whether to include unmodified nodes in the target development environment.
 86        environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
 87        default_start: The default plan start to use if not specified.
 88        default_end: The default plan end to use if not specified.
 89        enable_preview: Whether to enable preview for forward-only models in development environments.
 90        preview_start: The start time to use for forward-only previews. Defaults to the plan start.
 91        preview_min_intervals: The minimum number of intervals to preview for each forward-only preview snapshot.
 92        end_bounded: If set to true, the missing intervals will be bounded by the target end date, disregarding lookback,
 93            allow_partials, and other attributes that could cause the intervals to exceed the target end date.
 94        ensure_finalized_snapshots: Whether to compare against snapshots from the latest finalized
 95            environment state, or to use whatever snapshots are in the current environment state even if
 96            the environment is not finalized.
 97        start_override_per_model: A mapping of model FQNs to target start dates.
 98        end_override_per_model: A mapping of model FQNs to target end dates.
 99        ignore_cron: Whether to ignore the node's cron schedule when computing missing intervals.
100        explain: Whether to explain the plan instead of applying it.
101    """
102
103    def __init__(
104        self,
105        context_diff: ContextDiff,
106        start: t.Optional[TimeLike] = None,
107        end: t.Optional[TimeLike] = None,
108        execution_time: t.Optional[TimeLike] = None,
109        apply: t.Optional[t.Callable[[Plan], None]] = None,
110        restate_models: t.Optional[t.Iterable[str]] = None,
111        restate_all_snapshots: bool = False,
112        backfill_models: t.Optional[t.Iterable[str]] = None,
113        no_gaps: bool = False,
114        skip_backfill: bool = False,
115        empty_backfill: bool = False,
116        is_dev: bool = False,
117        forward_only: bool = False,
118        allow_destructive_models: t.Optional[t.Iterable[str]] = None,
119        allow_additive_models: t.Optional[t.Iterable[str]] = None,
120        environment_ttl: t.Optional[str] = None,
121        environment_suffix_target: EnvironmentSuffixTarget = EnvironmentSuffixTarget.default,
122        environment_catalog_mapping: t.Optional[t.Dict[re.Pattern, str]] = None,
123        categorizer_config: t.Optional[CategorizerConfig] = None,
124        auto_categorization_enabled: bool = True,
125        effective_from: t.Optional[TimeLike] = None,
126        include_unmodified: bool = False,
127        default_start: t.Optional[TimeLike] = None,
128        default_end: t.Optional[TimeLike] = None,
129        enable_preview: bool = False,
130        preview_start: t.Optional[TimeLike] = None,
131        preview_min_intervals: int = 0,
132        end_bounded: bool = False,
133        ensure_finalized_snapshots: bool = False,
134        explain: bool = False,
135        ignore_cron: bool = False,
136        start_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
137        end_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
138        console: t.Optional[PlanBuilderConsole] = None,
139        user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None,
140        selected_models: t.Optional[t.Set[str]] = None,
141    ):
142        self._context_diff = context_diff
143        self._no_gaps = no_gaps
144        self._skip_backfill = skip_backfill
145        self._empty_backfill = empty_backfill
146        self._is_dev = is_dev
147        self._forward_only = forward_only
148        self._allow_destructive_models = set(
149            allow_destructive_models if allow_destructive_models is not None else []
150        )
151        self._allow_additive_models = set(
152            allow_additive_models if allow_additive_models is not None else []
153        )
154        self._enable_preview = enable_preview
155        self._preview_start_provided = preview_start is not None
156        self._preview_start = preview_start
157        self._preview_min_intervals = preview_min_intervals
158        self._end_bounded = end_bounded
159        self._ensure_finalized_snapshots = ensure_finalized_snapshots
160        self._ignore_cron = ignore_cron
161        self._start_override_per_model = start_override_per_model
162        self._end_override_per_model = end_override_per_model
163        self._environment_ttl = environment_ttl
164        self._categorizer_config = categorizer_config or CategorizerConfig()
165        self._auto_categorization_enabled = auto_categorization_enabled
166        self._include_unmodified = include_unmodified
167        self._restate_models = set(restate_models) if restate_models is not None else None
168        self._restate_all_snapshots = restate_all_snapshots
169        self._effective_from = effective_from
170
171        # note: this deliberately doesnt default to now() here.
172        # There may be an significant delay between the PlanBuilder producing a Plan and the Plan actually being run
173        # so if execution_time=None is passed to the PlanBuilder, then the resulting Plan should also have execution_time=None
174        # in order to prevent the Plan that was intended to run "as at now" from having "now" fixed to some time in the past
175        # ref: https://github.com/SQLMesh/sqlmesh/pull/4702#discussion_r2140696156
176        self._execution_time = execution_time
177
178        self._backfill_models = backfill_models
179        self._end = end or default_end
180        self._default_start = default_start
181        self._apply = apply
182        self._console = console or get_console()
183        self._choices: t.Dict[SnapshotId, SnapshotChangeCategory] = {}
184        self._user_provided_flags = user_provided_flags
185        self._selected_models = selected_models
186        self._explain = explain
187
188        self._start = start
189        if not self._start and self._forward_only_preview_needed:
190            self._preview_start = self._preview_start or default_start or yesterday_ds()
191            # If a separate preview start was provided, don't let it shorten the
192            # plan start for regular backfills. Fallback preview starts preserve
193            # the previous preview behavior of using default_start or yesterday.
194            if self._preview_start_provided and not self._skip_backfill:
195                self._start = default_start or yesterday_ds()
196            else:
197                self._start = self._preview_start
198
199        if not self._start and self._non_forward_only_preview_needed:
200            self._start = default_start or yesterday_ds()
201
202        self._plan_id: str = random_id()
203        self._model_fqn_to_snapshot = {s.name: s for s in self._context_diff.snapshots.values()}
204
205        self.override_start = start is not None
206        self.override_end = end is not None
207        self.environment_naming_info = EnvironmentNamingInfo.from_environment_catalog_mapping(
208            environment_catalog_mapping or {},
209            name=self._context_diff.environment,
210            suffix_target=environment_suffix_target,
211            normalize_name=self._context_diff.normalize_environment_name,
212            gateway_managed=self._context_diff.gateway_managed_virtual_layer,
213        )
214
215        self._latest_plan: t.Optional[Plan] = None
216
217    @property
218    def is_start_and_end_allowed(self) -> bool:
219        """Indicates whether this plan allows to set the start and end dates."""
220        return self._is_dev or bool(self._restate_models)
221
222    @property
223    def start(self) -> t.Optional[TimeLike]:
224        if self._start and is_relative(self._start):
225            # only do this for relative expressions otherwise inclusive date strings like '2020-01-01' can be turned into exclusive timestamps eg '2020-01-01 00:00:00'
226            return to_datetime(self._start, relative_base=to_datetime(self.execution_time))
227        return self._start
228
229    @property
230    def end(self) -> t.Optional[TimeLike]:
231        if self._end and is_relative(self._end):
232            # only do this for relative expressions otherwise inclusive date strings like '2020-01-01' can be turned into exclusive timestamps eg '2020-01-01 00:00:00'
233            return to_datetime(self._end, relative_base=to_datetime(self.execution_time))
234        return self._end
235
236    @cached_property
237    def execution_time(self) -> TimeLike:
238        # this is cached to return a stable value from now() in the places where the execution time matters for resolving relative date strings
239        # during the plan building process
240        return self._execution_time or now()
241
242    def set_start(self, new_start: TimeLike) -> PlanBuilder:
243        self._start = new_start
244        if not self._preview_start_provided and self._forward_only_preview_needed:
245            self._preview_start = new_start
246        self.override_start = True
247        self._latest_plan = None
248        return self
249
250    def set_end(self, new_end: TimeLike) -> PlanBuilder:
251        self._end = new_end
252        self.override_end = True
253        self._latest_plan = None
254        return self
255
256    def set_effective_from(self, effective_from: t.Optional[TimeLike]) -> PlanBuilder:
257        """Sets the effective date for all new snapshots in the plan.
258
259        Note: this is only applicable for forward-only plans.
260
261        Args:
262            effective_from: The effective date to set.
263        """
264        self._effective_from = effective_from
265        if effective_from and self._is_dev and not self.override_start:
266            self._start = effective_from
267            if not self._preview_start_provided and self._forward_only_preview_needed:
268                self._preview_start = effective_from
269        self._latest_plan = None
270        return self
271
272    def set_choice(self, snapshot: Snapshot, choice: SnapshotChangeCategory) -> PlanBuilder:
273        """Sets a snapshot version based on the user choice.
274
275        Args:
276            snapshot: The target snapshot.
277            choice: The user decision on how to version the target snapshot and its children.
278        """
279        if not self._is_new_snapshot(snapshot):
280            raise PlanError(
281                f"A choice can't be changed for the existing version of {snapshot.name}."
282            )
283        if (
284            not self._context_diff.directly_modified(snapshot.name)
285            and snapshot.snapshot_id not in self._context_diff.added
286        ):
287            raise PlanError(f"Only directly modified models can be categorized ({snapshot.name}).")
288
289        self._choices[snapshot.snapshot_id] = choice
290        self._latest_plan = None
291        return self
292
293    def apply(self) -> None:
294        """Builds and applies the plan."""
295        if not self._apply:
296            raise PlanError("Plan was not initialized with an applier.")
297        self._apply(self.build())
298
299    def build(self) -> Plan:
300        """Builds the plan."""
301        if self._latest_plan:
302            return self._latest_plan
303
304        self._ensure_new_env_with_changes()
305        self._ensure_valid_date_range()
306        self._ensure_no_broken_references()
307
308        self._apply_effective_from()
309
310        dag = self._build_dag()
311        directly_modified, indirectly_modified = self._build_directly_and_indirectly_modified(dag)
312
313        self._check_destructive_additive_changes(directly_modified)
314        self._categorize_snapshots(dag, indirectly_modified)
315        self._adjust_snapshot_intervals()
316
317        deployability_index = (
318            DeployabilityIndex.create(
319                self._context_diff.snapshots.values(),
320                start=self._start,
321                start_override_per_model=self._start_override_per_model,
322            )
323            if self._is_dev
324            else DeployabilityIndex.all_deployable()
325        )
326
327        restatements = self._build_restatements(
328            dag,
329            earliest_interval_start(self._context_diff.snapshots.values(), self.execution_time),
330        )
331        models_to_backfill = self._build_models_to_backfill(dag, restatements)
332
333        end_override_per_model = self._end_override_per_model
334        if end_override_per_model and self.override_end:
335            # If the end date was provided explicitly by a user, then interval end for each individual
336            # model should be ignored.
337            end_override_per_model = None
338
339        # this deliberately uses the passed in self._execution_time and not self.execution_time cached property
340        # the reason is because that there can be a delay between the Plan being built and the Plan being actually run,
341        # so this ensures that an _execution_time of None can be propagated to the Plan and thus be re-resolved to
342        # the current timestamp of when the Plan is eventually run
343        plan_execution_time = self._execution_time
344
345        plan = Plan(
346            context_diff=self._context_diff,
347            plan_id=self._plan_id,
348            provided_start=self.start,
349            provided_end=self.end,
350            is_dev=self._is_dev,
351            skip_backfill=self._skip_backfill,
352            empty_backfill=self._empty_backfill,
353            no_gaps=self._no_gaps,
354            forward_only=self._forward_only,
355            explain=self._explain,
356            allow_destructive_models=t.cast(t.Set, self._allow_destructive_models),
357            allow_additive_models=t.cast(t.Set, self._allow_additive_models),
358            include_unmodified=self._include_unmodified,
359            environment_ttl=self._environment_ttl,
360            environment_naming_info=self.environment_naming_info,
361            directly_modified=directly_modified,
362            indirectly_modified=indirectly_modified,
363            deployability_index=deployability_index,
364            selected_models_to_restate=self._restate_models,
365            restatements=restatements,
366            restate_all_snapshots=self._restate_all_snapshots,
367            start_override_per_model=self._start_override_per_model,
368            end_override_per_model=end_override_per_model,
369            selected_models_to_backfill=self._backfill_models,
370            models_to_backfill=models_to_backfill,
371            effective_from=self._effective_from,
372            execution_time=plan_execution_time,
373            end_bounded=self._end_bounded,
374            ensure_finalized_snapshots=self._ensure_finalized_snapshots,
375            ignore_cron=self._ignore_cron,
376            user_provided_flags=self._user_provided_flags,
377            selected_models=self._selected_models,
378        )
379        self._latest_plan = plan
380        return plan
381
382    def _build_dag(self) -> DAG[SnapshotId]:
383        dag: DAG[SnapshotId] = DAG()
384        for s_id, context_snapshot in self._context_diff.snapshots.items():
385            dag.add(s_id, context_snapshot.parents)
386        return dag
387
388    def _build_restatements(
389        self, dag: DAG[SnapshotId], earliest_interval_start: TimeLike
390    ) -> t.Dict[SnapshotId, Interval]:
391        restate_models = self._restate_models
392        if restate_models == set():
393            # This is a warning but we print this as error since the Console is lacking API for warnings.
394            self._console.log_error(
395                "Provided restated models do not match any models. No models will be included in plan."
396            )
397            return {}
398
399        restatements: t.Dict[SnapshotId, Interval] = {}
400        forward_only_preview_needed = self._forward_only_preview_needed
401        is_preview = False
402        if not restate_models and forward_only_preview_needed:
403            # Add model names for new forward-only snapshots to the restatement list
404            # in order to compute previews.
405            restate_models = {
406                s.name
407                for s in self._context_diff.new_snapshots.values()
408                if s.is_model
409                and not s.is_symbolic
410                and (s.is_forward_only or s.model.forward_only)
411                and not s.is_no_preview
412                and (
413                    # Metadata changes should not be previewed.
414                    self._context_diff.directly_modified(s.name)
415                    or self._context_diff.indirectly_modified(s.name)
416                )
417            }
418            is_preview = True
419
420        if not restate_models:
421            return {}
422
423        start = self._start or earliest_interval_start
424        end = self._end or now()
425
426        # Add restate snapshots and their downstream snapshots
427        for model_fqn in restate_models:
428            if model_fqn not in self._model_fqn_to_snapshot:
429                raise PlanError(f"Cannot restate model '{model_fqn}'. Model does not exist.")
430
431        # Get restatement intervals for all restated snapshots and make sure that if an incremental snapshot expands it's
432        # restatement range that it's downstream dependencies all expand their restatement ranges as well.
433        for s_id in dag:
434            snapshot = self._context_diff.snapshots[s_id]
435
436            if is_preview and snapshot.is_no_preview:
437                continue
438
439            # Since we are traversing the graph in topological order and the largest interval range is pushed down
440            # the graph we just have to check our immediate parents in the graph and not the whole upstream graph.
441            restating_parents = [
442                self._context_diff.snapshots[s] for s in snapshot.parents if s in restatements
443            ]
444
445            if not restating_parents and snapshot.name not in restate_models:
446                continue
447
448            if not forward_only_preview_needed:
449                if self._is_dev and not snapshot.is_paused:
450                    self._console.log_warning(
451                        f"Cannot restate model '{snapshot.name}' because the current version is used in production. "
452                        "Run the restatement against the production environment instead to restate this model."
453                    )
454                    continue
455                elif (not self._is_dev or not snapshot.is_paused) and snapshot.disable_restatement:
456                    self._console.log_warning(
457                        f"Cannot restate model '{snapshot.name}'. "
458                        "Restatement is disabled for this model to prevent possible data loss. "
459                        "If you want to restate this model, change the model's `disable_restatement` setting to `false`."
460                    )
461                    continue
462                elif snapshot.is_seed:
463                    logger.info("Skipping restatement for model '%s'", snapshot.name)
464                    continue
465
466            possible_intervals = {
467                restatements[p.snapshot_id] for p in restating_parents if p.is_incremental
468            }
469            removal_start = (
470                self._forward_only_preview_start(snapshot, start, end) if is_preview else start
471            )
472            possible_intervals.add(
473                snapshot.get_removal_interval(
474                    removal_start,
475                    end,
476                    self._execution_time,
477                    strict=False,
478                    is_preview=is_preview,
479                )
480            )
481            snapshot_start = min(i[0] for i in possible_intervals)
482            snapshot_end = max(i[1] for i in possible_intervals)
483
484            # We may be tasked with restating a time range smaller than the target snapshot interval unit
485            # For example, restating an hour of Hourly Model A, which has a downstream dependency of Daily Model B
486            # we need to ensure the whole affected day in Model B is restated
487            floored_snapshot_start = snapshot.node.interval_unit.cron_floor(snapshot_start)
488            floored_snapshot_end = snapshot.node.interval_unit.cron_floor(snapshot_end)
489            if to_timestamp(floored_snapshot_end) < snapshot_end:
490                snapshot_start = to_timestamp(floored_snapshot_start)
491                snapshot_end = to_timestamp(
492                    snapshot.node.interval_unit.cron_next(floored_snapshot_end)
493                )
494
495            restatements[s_id] = (snapshot_start, snapshot_end)
496
497        return restatements
498
499    def _forward_only_preview_start(
500        self, snapshot: Snapshot, default_start: TimeLike, end: TimeLike
501    ) -> TimeLike:
502        preview_start = self._preview_start or default_start
503        if not self._preview_min_intervals:
504            return preview_start
505
506        relative_base = to_datetime(self.execution_time)
507        preview_end = to_datetime(end, relative_base=relative_base)
508        min_start = snapshot.node.cron_floor(preview_end)
509        for _ in range(self._preview_min_intervals):
510            min_start = snapshot.node.cron_prev(min_start)
511
512        return min(to_datetime(preview_start, relative_base=relative_base), min_start)
513
514    def _build_directly_and_indirectly_modified(
515        self, dag: DAG[SnapshotId]
516    ) -> t.Tuple[t.Set[SnapshotId], SnapshotMapping]:
517        """Builds collections of directly and indirectly modified snapshots.
518
519        Returns:
520            The tuple in which the first element contains a list of added and directly modified
521            snapshots while the second element contains a mapping of indirectly modified snapshots.
522        """
523        directly_modified = set()
524        all_indirectly_modified = set()
525
526        for s_id in dag:
527            if s_id.name in self._context_diff.modified_snapshots:
528                if self._context_diff.directly_modified(s_id.name):
529                    directly_modified.add(s_id)
530                else:
531                    all_indirectly_modified.add(s_id)
532            elif s_id in self._context_diff.added:
533                directly_modified.add(s_id)
534
535        indirectly_modified: SnapshotMapping = defaultdict(set)
536        for snapshot in directly_modified:
537            for downstream_s_id in dag.downstream(snapshot.snapshot_id):
538                if downstream_s_id in all_indirectly_modified:
539                    indirectly_modified[snapshot.snapshot_id].add(downstream_s_id)
540
541        return (
542            directly_modified,
543            indirectly_modified,
544        )
545
546    def _build_models_to_backfill(
547        self, dag: DAG[SnapshotId], restatements: t.Collection[SnapshotId]
548    ) -> t.Optional[t.Set[str]]:
549        backfill_models = (
550            self._backfill_models
551            if self._backfill_models is not None
552            else [r.name for r in restatements]
553            # Only backfill models explicitly marked for restatement.
554            if self._restate_models
555            else None
556        )
557        if backfill_models is None:
558            return None
559        return {
560            self._context_diff.snapshots[s_id].name
561            for s_id in dag.subdag(
562                *[
563                    self._model_fqn_to_snapshot[m].snapshot_id
564                    for m in backfill_models
565                    if m in self._model_fqn_to_snapshot
566                ]
567            ).sorted
568        }
569
570    def _adjust_snapshot_intervals(self) -> None:
571        for new, old in self._context_diff.modified_snapshots.values():
572            if not new.is_model or not old.is_model:
573                continue
574            is_same_version = old.version_get_or_generate() == new.version_get_or_generate()
575            if is_same_version and should_force_rebuild(old, new):
576                # If the difference between 2 snapshots requires a full rebuild,
577                # then clear the intervals for the new snapshot.
578                self._context_diff.snapshots[new.snapshot_id].intervals = []
579            elif new.snapshot_id in self._context_diff.new_snapshots:
580                new.intervals = []
581                new.dev_intervals = []
582                if is_same_version:
583                    new.merge_intervals(old)
584                    if new.is_forward_only:
585                        new.dev_intervals = new.intervals.copy()
586
587    def _check_destructive_additive_changes(self, directly_modified: t.Set[SnapshotId]) -> None:
588        for s_id in sorted(directly_modified):
589            if s_id.name not in self._context_diff.modified_snapshots:
590                continue
591
592            snapshot = self._context_diff.snapshots[s_id]
593            needs_destructive_check = snapshot.needs_destructive_check(
594                self._allow_destructive_models
595            )
596            needs_additive_check = snapshot.needs_additive_check(self._allow_additive_models)
597            # should we raise/warn if this snapshot has/inherits a destructive change?
598            should_raise_or_warn = (self._is_forward_only_change(s_id) or self._forward_only) and (
599                needs_destructive_check or needs_additive_check
600            )
601
602            if not should_raise_or_warn or not snapshot.is_model:
603                continue
604
605            new, old = self._context_diff.modified_snapshots[snapshot.name]
606
607            # we must know all columns_to_types to determine whether a change is destructive
608            old_columns_to_types = old.model.columns_to_types or {}
609            new_columns_to_types = new.model.columns_to_types or {}
610
611            if columns_to_types_all_known(old_columns_to_types) and columns_to_types_all_known(
612                new_columns_to_types
613            ):
614                alter_operations = t.cast(
615                    t.List[TableAlterOperation],
616                    get_schema_differ(snapshot.model.dialect).compare_columns(
617                        new.name,
618                        old_columns_to_types,
619                        new_columns_to_types,
620                        ignore_destructive=new.model.on_destructive_change.is_ignore,
621                        ignore_additive=new.model.on_additive_change.is_ignore,
622                    ),
623                )
624
625                snapshot_name = snapshot.name
626                model_dialect = snapshot.model.dialect
627
628                if needs_destructive_check and has_drop_alteration(alter_operations):
629                    self._console.log_destructive_change(
630                        snapshot_name,
631                        alter_operations,
632                        model_dialect,
633                        error=not snapshot.model.on_destructive_change.is_warn,
634                    )
635                    if snapshot.model.on_destructive_change.is_error:
636                        raise PlanError(
637                            "Plan requires a destructive change to a forward-only model."
638                        )
639
640                if needs_additive_check and has_additive_alteration(alter_operations):
641                    self._console.log_additive_change(
642                        snapshot_name,
643                        alter_operations,
644                        model_dialect,
645                        error=not snapshot.model.on_additive_change.is_warn,
646                    )
647                    if snapshot.model.on_additive_change.is_error:
648                        raise PlanError("Plan requires an additive change to a forward-only model.")
649
650    def _categorize_snapshots(
651        self, dag: DAG[SnapshotId], indirectly_modified: SnapshotMapping
652    ) -> None:
653        """Automatically categorizes snapshots that can be automatically categorized and
654        returns a list of added and directly modified snapshots as well as the mapping of
655        indirectly modified snapshots.
656        """
657
658        # Iterating in DAG order since a category for a snapshot may depend on the categories
659        # assigned to its upstream dependencies.
660        for s_id in dag:
661            snapshot = self._context_diff.snapshots.get(s_id)
662
663            if not snapshot or not self._is_new_snapshot(snapshot):
664                continue
665
666            forward_only = self._forward_only or self._is_forward_only_change(s_id)
667            if forward_only and s_id.name in self._context_diff.modified_snapshots:
668                new, old = self._context_diff.modified_snapshots[s_id.name]
669                if is_breaking_kind_change(old, new) or snapshot.is_seed:
670                    # Breaking kind changes and seed changes can't be forward-only.
671                    forward_only = False
672
673            if s_id in self._choices:
674                snapshot.categorize_as(self._choices[s_id], forward_only)
675                continue
676
677            if s_id in self._context_diff.added:
678                snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only)
679            elif s_id.name in self._context_diff.modified_snapshots:
680                self._categorize_snapshot(snapshot, forward_only, dag, indirectly_modified)
681
682    def _categorize_snapshot(
683        self,
684        snapshot: Snapshot,
685        forward_only: bool,
686        dag: DAG[SnapshotId],
687        indirectly_modified: SnapshotMapping,
688    ) -> None:
689        s_id = snapshot.snapshot_id
690
691        if self._context_diff.directly_modified(s_id.name):
692            if self._auto_categorization_enabled:
693                new, old = self._context_diff.modified_snapshots[s_id.name]
694                if is_breaking_kind_change(old, new):
695                    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, False)
696                    return
697
698                s_id_with_missing_columns: t.Optional[SnapshotId] = None
699                this_sid_with_downstream = indirectly_modified.get(s_id, set()) | {s_id}
700                for downstream_s_id in this_sid_with_downstream:
701                    downstream_snapshot = self._context_diff.snapshots[downstream_s_id]
702                    if (
703                        downstream_snapshot.is_model
704                        and downstream_snapshot.model.columns_to_types is None
705                    ):
706                        s_id_with_missing_columns = downstream_s_id
707                        break
708
709                if s_id_with_missing_columns is None:
710                    change_category = categorize_change(new, old, config=self._categorizer_config)
711                    if change_category is not None:
712                        snapshot.categorize_as(change_category, forward_only)
713                else:
714                    mode = self._categorizer_config.dict().get(
715                        new.model.source_type, AutoCategorizationMode.OFF
716                    )
717                    if mode == AutoCategorizationMode.FULL:
718                        snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only)
719        elif self._context_diff.indirectly_modified(snapshot.name):
720            if snapshot.is_materialized_view and not forward_only:
721                # We categorize changes as breaking to allow for instantaneous switches in a virtual layer.
722                # Otherwise, there might be a potentially long downtime during MVs recreation.
723                # In the case of forward-only changes this optimization is not applicable because we want to continue
724                # using the same (existing) table version.
725                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING, forward_only)
726                return
727
728            all_upstream_forward_only = set()
729            all_upstream_categories = set()
730            direct_parent_categories = set()
731
732            for p_id in dag.upstream(s_id):
733                parent = self._context_diff.snapshots.get(p_id)
734
735                if parent and self._is_new_snapshot(parent):
736                    all_upstream_categories.add(parent.change_category)
737                    all_upstream_forward_only.add(parent.is_forward_only)
738                    if p_id in snapshot.parents:
739                        direct_parent_categories.add(parent.change_category)
740
741            if all_upstream_forward_only == {True} or (
742                snapshot.is_model and snapshot.model.forward_only
743            ):
744                forward_only = True
745
746            if direct_parent_categories.intersection(
747                {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
748            ):
749                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING, forward_only)
750            elif not direct_parent_categories:
751                snapshot.categorize_as(
752                    self._get_orphaned_indirect_change_category(snapshot), forward_only
753                )
754            elif all_upstream_categories == {SnapshotChangeCategory.METADATA}:
755                snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only)
756            else:
757                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING, forward_only)
758        else:
759            # Metadata updated.
760            snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only)
761
762    def _get_orphaned_indirect_change_category(
763        self, indirect_snapshot: Snapshot
764    ) -> SnapshotChangeCategory:
765        """Sometimes an indirectly changed downstream snapshot ends up with no directly changed parents introduced in the same plan.
766        This may happen when 2 or more parent models were changed independently in different plans and then the changes were
767        merged together and applied in a single plan. As a result, a combination of 2 or more previously changed parents produces
768        a new downstream snapshot not previously seen.
769
770        This function is used to infer the correct change category for such downstream snapshots based on change categories of their parents.
771        """
772        previous_snapshot = self._context_diff.modified_snapshots[indirect_snapshot.name][1]
773        previous_parent_snapshot_ids = {p.name: p for p in previous_snapshot.parents}
774
775        current_parent_snapshots = [
776            self._context_diff.snapshots[p_id]
777            for p_id in indirect_snapshot.parents
778            if p_id in self._context_diff.snapshots
779        ]
780
781        indirect_category: t.Optional[SnapshotChangeCategory] = None
782        for current_parent_snapshot in current_parent_snapshots:
783            if current_parent_snapshot.name not in previous_parent_snapshot_ids:
784                # This is a new parent so falling back to INDIRECT_BREAKING
785                return SnapshotChangeCategory.INDIRECT_BREAKING
786            pevious_parent_snapshot_id = previous_parent_snapshot_ids[current_parent_snapshot.name]
787
788            if current_parent_snapshot.snapshot_id == pevious_parent_snapshot_id:
789                # There were no new versions of this parent since the previous version of this snapshot,
790                # so we can skip it
791                continue
792
793            # Find the previous snapshot ID of the same parent in the historical chain
794            previous_parent_found = False
795            previous_parent_categories = set()
796            for pv in reversed(current_parent_snapshot.all_versions):
797                pv_snapshot_id = pv.snapshot_id(current_parent_snapshot.name)
798                if pv_snapshot_id == pevious_parent_snapshot_id:
799                    previous_parent_found = True
800                    break
801                previous_parent_categories.add(pv.change_category)
802
803            if not previous_parent_found:
804                # The previous parent is not in the historical chain so falling back to INDIRECT_BREAKING
805                return SnapshotChangeCategory.INDIRECT_BREAKING
806
807            if previous_parent_categories.intersection(
808                {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
809            ):
810                # One of the new parents in the chain was breaking so this indirect snapshot is breaking
811                return SnapshotChangeCategory.INDIRECT_BREAKING
812
813            if previous_parent_categories.intersection(
814                {
815                    SnapshotChangeCategory.NON_BREAKING,
816                    SnapshotChangeCategory.INDIRECT_NON_BREAKING,
817                }
818            ):
819                # All changes in the chain were non-breaking so this indirect snapshot can be non-breaking too
820                indirect_category = SnapshotChangeCategory.INDIRECT_NON_BREAKING
821            elif (
822                previous_parent_categories == {SnapshotChangeCategory.METADATA}
823                and indirect_category is None
824            ):
825                # All changes in the chain were metadata so this indirect snapshot can be metadata too
826                indirect_category = SnapshotChangeCategory.METADATA
827
828        return indirect_category or SnapshotChangeCategory.INDIRECT_BREAKING
829
830    def _apply_effective_from(self) -> None:
831        if self._effective_from:
832            if not self._forward_only:
833                raise PlanError("Effective date can only be set for a forward-only plan.")
834            if to_datetime(self._effective_from) > now():
835                raise PlanError("Effective date cannot be in the future.")
836
837        for snapshot in self._context_diff.new_snapshots.values():
838            if (
839                snapshot.evaluatable
840                and not snapshot.disable_restatement
841                and (not snapshot.full_history_restatement_only or not snapshot.is_incremental)
842            ):
843                snapshot.effective_from = self._effective_from
844
845    def _is_forward_only_change(self, s_id: SnapshotId) -> bool:
846        if not self._context_diff.directly_modified(
847            s_id.name
848        ) and not self._context_diff.indirectly_modified(s_id.name):
849            return False
850        snapshot = self._context_diff.snapshots[s_id]
851        if snapshot.name in self._context_diff.modified_snapshots:
852            _, old = self._context_diff.modified_snapshots[snapshot.name]
853            # If the model kind has changed in a breaking way, then we can't consider this to be a forward-only change.
854            if snapshot.is_model and is_breaking_kind_change(old, snapshot):
855                return False
856        return (
857            snapshot.is_model and snapshot.model.forward_only and bool(snapshot.previous_versions)
858        )
859
860    def _is_new_snapshot(self, snapshot: Snapshot) -> bool:
861        """Returns True if the given snapshot is a new snapshot in this plan."""
862        return snapshot.snapshot_id in self._context_diff.new_snapshots
863
864    def _ensure_valid_date_range(self) -> None:
865        if (self.override_start or self.override_end) and not self.is_start_and_end_allowed:
866            raise PlanError(
867                "The start and end dates can't be set for a production plan without restatements."
868            )
869
870        if (start := self.start) and (end := self.end):
871            if to_datetime(start) > to_datetime(end):
872                raise PlanError(
873                    f"Plan end date: '{time_like_to_str(end)}' must be after the plan start date: '{time_like_to_str(start)}'"
874                )
875
876        if end := self.end:
877            if to_datetime(end) > to_datetime(self.execution_time):
878                raise PlanError(
879                    f"Plan end date: '{time_like_to_str(end)}' cannot be in the future (execution time: '{time_like_to_str(self.execution_time)}')"
880                )
881
882        # Validate model-specific start/end dates
883        if (start := self.start or self._default_start) and (end := self.end):
884            start_ts = to_datetime(start)
885            end_ts = to_datetime(end)
886            if start_ts > end_ts:
887                models_to_check: t.Set[str] = (
888                    set(self._backfill_models or [])
889                    | set(self._context_diff.modified_snapshots.keys())
890                    | {s.name for s in self._context_diff.added}
891                    | set((self._end_override_per_model or {}).keys())
892                )
893                for model_name in models_to_check:
894                    if snapshot := self._model_fqn_to_snapshot.get(model_name):
895                        if snapshot.node.start is None or to_datetime(snapshot.node.start) > end_ts:
896                            raise PlanError(
897                                f"Model '{model_name}': Start date / time '({time_like_to_str(start_ts)})' can't be greater than end date / time '({time_like_to_str(end_ts)})'.\n"
898                                f"Set the `start` attribute in your project config model defaults to avoid this issue."
899                            )
900
901    def _ensure_no_broken_references(self) -> None:
902        for snapshot in self._context_diff.snapshots.values():
903            broken_references = {
904                x.name for x in self._context_diff.removed_snapshots.values() if not x.is_external
905            } & {x for x in snapshot.node.depends_on}
906            if broken_references:
907                broken_references_msg = ", ".join(f"'{x}'" for x in broken_references)
908                raise PlanError(
909                    f"""Removed {broken_references_msg} are referenced in '{snapshot.name}'. Please remove broken references before proceeding."""
910                )
911
912    def _ensure_new_env_with_changes(self) -> None:
913        if (
914            self._is_dev
915            and not self._include_unmodified
916            and self._context_diff.is_new_environment
917            and not self._context_diff.has_snapshot_changes
918            and not self._context_diff.has_environment_statements_changes
919            and not self._backfill_models
920        ):
921            raise NoChangesPlanError(
922                f"Creating a new environment requires a change, but project files match the `{self._context_diff.create_from}` environment. Make a change or use the --include-unmodified flag to create a new environment without changes."
923            )
924
925    @cached_property
926    def _forward_only_preview_needed(self) -> bool:
927        """Determines whether the plan should compute previews for forward-only changes (if there are any)."""
928        return self._is_dev and (
929            self._forward_only
930            or (
931                self._enable_preview
932                and any(
933                    snapshot.model.forward_only
934                    for snapshot in self._modified_and_added_snapshots
935                    if snapshot.is_model
936                )
937            )
938        )
939
940    @cached_property
941    def _non_forward_only_preview_needed(self) -> bool:
942        if not self._is_dev:
943            return False
944        for snapshot in self._modified_and_added_snapshots:
945            if not snapshot.is_model:
946                continue
947            if (
948                not snapshot.virtual_environment_mode.is_full
949                or snapshot.model.auto_restatement_cron is not None
950            ):
951                return True
952        return False
953
954    @cached_property
955    def _modified_and_added_snapshots(self) -> t.List[Snapshot]:
956        return [
957            snapshot
958            for snapshot in self._context_diff.snapshots.values()
959            if snapshot.name in self._context_diff.modified_snapshots
960            or snapshot.snapshot_id in self._context_diff.added
961        ]