1"""
   2# SnapshotEvaluator
   3
   4A snapshot evaluator is responsible for evaluating a snapshot given some runtime arguments, e.g. start
   5and end timestamps.
   6
   7# Evaluation
   8
   9Snapshot evaluation involves determining the queries necessary to evaluate a snapshot and using
  10`sqlmesh.core.engine_adapter` to execute the queries. Schemas, tables, and views are created if
  11they don't exist and data is inserted when applicable.
  12
  13A snapshot evaluator also promotes and demotes snapshots to a given environment.
  14
  15# Audits
  16
  17A snapshot evaluator can also run the audits for a snapshot's node. This is often done after a snapshot
  18has been evaluated to check for data quality issues.
  19
  20For more information about audits, see `sqlmesh.core.audit`.
  21"""
  22
  23from __future__ import annotations
  24
  25import abc
  26import logging
  27import typing as t
  28import sys
  29from collections import defaultdict
  30from contextlib import contextmanager
  31from functools import reduce
  32
  33from sqlglot import exp, select
  34from sqlglot.executor import execute
  35from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_not_exception_type
  36
  37from sqlmesh.core import constants as c
  38from sqlmesh.core import dialect as d
  39from sqlmesh.core.audit import Audit, StandaloneAudit
  40from sqlmesh.core.dialect import schema_
  41from sqlmesh.core.engine_adapter.shared import InsertOverwriteStrategy, DataObjectType, DataObject
  42from sqlmesh.core.model.meta import GrantsTargetLayer
  43from sqlmesh.core.macros import RuntimeStage
  44from sqlmesh.core.model import (
  45    AuditResult,
  46    IncrementalUnmanagedKind,
  47    Model,
  48    SeedModel,
  49    SCDType2ByColumnKind,
  50    SCDType2ByTimeKind,
  51    ViewKind,
  52    CustomKind,
  53)
  54from sqlmesh.core.model.kind import _Incremental, DbtCustomKind
  55from sqlmesh.utils import CompletionStatus, columns_to_types_all_known
  56from sqlmesh.core.schema_diff import (
  57    has_drop_alteration,
  58    TableAlterOperation,
  59    has_additive_alteration,
  60)
  61from sqlmesh.core.snapshot import (
  62    DeployabilityIndex,
  63    Intervals,
  64    Snapshot,
  65    SnapshotId,
  66    SnapshotIdBatch,
  67    SnapshotInfoLike,
  68    SnapshotTableCleanupTask,
  69)
  70from sqlmesh.core.snapshot.execution_tracker import QueryExecutionTracker
  71from sqlmesh.utils import random_id, CorrelationId, AttributeDict
  72from sqlmesh.utils.concurrency import (
  73    concurrent_apply_to_snapshots,
  74    concurrent_apply_to_values,
  75    NodeExecutionFailedError,
  76)
  77from sqlmesh.utils.date import TimeLike, now, time_like_to_str
  78from sqlmesh.utils.errors import (
  79    ConfigError,
  80    DestructiveChangeError,
  81    MigrationNotSupportedError,
  82    SQLMeshError,
  83    format_destructive_change_msg,
  84    format_additive_change_msg,
  85    AdditiveChangeError,
  86)
  87from sqlmesh.utils.jinja import MacroReturnVal
  88
  89if sys.version_info >= (3, 12):
  90    from importlib import metadata
  91else:
  92    import importlib_metadata as metadata  # type: ignore
  93
  94if t.TYPE_CHECKING:
  95    from sqlmesh.core.engine_adapter._typing import DF, QueryOrDF
  96    from sqlmesh.core.engine_adapter.base import EngineAdapter
  97    from sqlmesh.core.environment import EnvironmentNamingInfo
  98
  99logger = logging.getLogger(__name__)
 100
 101
 102class SnapshotCreationFailedError(SQLMeshError):
 103    def __init__(
 104        self, errors: t.List[NodeExecutionFailedError[SnapshotId]], skipped: t.List[SnapshotId]
 105    ):
 106        messages = "\n\n".join(f"{error}\n  {error.__cause__}" for error in errors)
 107        super().__init__(f"Physical table creation failed:\n\n{messages}")
 108        self.errors = errors
 109        self.skipped = skipped
 110
 111
 112class SnapshotEvaluator:
 113    """Evaluates a snapshot given runtime arguments through an arbitrary EngineAdapter.
 114
 115    The SnapshotEvaluator contains the business logic to generically evaluate a snapshot.
 116    It is responsible for delegating queries to the EngineAdapter. The SnapshotEvaluator
 117    does not directly communicate with the underlying execution engine.
 118
 119    Args:
 120        adapters: A single EngineAdapter or a dictionary of EngineAdapters where
 121            the key is the gateway name. When a dictionary is provided, and not an
 122            explicit default gateway its first item is treated as the default
 123            adapter and used for the virtual layer.
 124        ddl_concurrent_tasks: The number of concurrent tasks used for DDL
 125            operations (table / view creation, deletion, etc). Default: 1.
 126    """
 127
 128    def __init__(
 129        self,
 130        adapters: EngineAdapter | t.Dict[str, EngineAdapter],
 131        ddl_concurrent_tasks: int = 1,
 132        selected_gateway: t.Optional[str] = None,
 133    ):
 134        self.adapters = (
 135            adapters if isinstance(adapters, t.Dict) else {selected_gateway or "": adapters}
 136        )
 137        self.execution_tracker = QueryExecutionTracker()
 138        self.adapters = {
 139            gateway: adapter.with_settings(query_execution_tracker=self.execution_tracker)
 140            for gateway, adapter in self.adapters.items()
 141        }
 142        self.adapter = (
 143            next(iter(self.adapters.values()))
 144            if not selected_gateway
 145            else self.adapters[selected_gateway]
 146        )
 147        self.selected_gateway = selected_gateway
 148        self.ddl_concurrent_tasks = ddl_concurrent_tasks
 149
 150    def evaluate(
 151        self,
 152        snapshot: Snapshot,
 153        *,
 154        start: TimeLike,
 155        end: TimeLike,
 156        execution_time: TimeLike,
 157        snapshots: t.Dict[str, Snapshot],
 158        allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
 159        allow_additive_snapshots: t.Optional[t.Set[str]] = None,
 160        deployability_index: t.Optional[DeployabilityIndex] = None,
 161        batch_index: int = 0,
 162        target_table_exists: t.Optional[bool] = None,
 163        **kwargs: t.Any,
 164    ) -> t.Optional[str]:
 165        """Renders the snapshot's model, executes it and stores the result in the snapshot's physical table.
 166
 167        Args:
 168            snapshot: Snapshot to evaluate.
 169            start: The start datetime to render.
 170            end: The end datetime to render.
 171            execution_time: The date/time time reference to use for execution time.
 172            snapshots: All upstream snapshots (by name) to use for expansion and mapping of physical locations.
 173            allow_destructive_snapshots: Snapshots for which destructive schema changes are allowed.
 174            allow_additive_snapshots: Snapshots for which additive schema changes are allowed.
 175            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 176            batch_index: If the snapshot is part of a batch of related snapshots; which index in the batch is it
 177            target_table_exists: Whether the target table exists. If None, the table will be checked for existence.
 178            kwargs: Additional kwargs to pass to the renderer.
 179
 180        Returns:
 181            The WAP ID of this evaluation if supported, None otherwise.
 182        """
 183        with self.execution_tracker.track_execution(
 184            SnapshotIdBatch(snapshot_id=snapshot.snapshot_id, batch_id=batch_index)
 185        ):
 186            result = self._evaluate_snapshot(
 187                start=start,
 188                end=end,
 189                execution_time=execution_time,
 190                snapshot=snapshot,
 191                snapshots=snapshots,
 192                allow_destructive_snapshots=allow_destructive_snapshots or set(),
 193                allow_additive_snapshots=allow_additive_snapshots or set(),
 194                deployability_index=deployability_index,
 195                batch_index=batch_index,
 196                target_table_exists=target_table_exists,
 197                **kwargs,
 198            )
 199        if result is None or isinstance(result, str):
 200            return result
 201        raise SQLMeshError(
 202            f"Unexpected result {result} when evaluating snapshot {snapshot.snapshot_id}."
 203        )
 204
 205    def evaluate_and_fetch(
 206        self,
 207        snapshot: Snapshot,
 208        *,
 209        start: TimeLike,
 210        end: TimeLike,
 211        execution_time: TimeLike,
 212        snapshots: t.Dict[str, Snapshot],
 213        limit: int,
 214        deployability_index: t.Optional[DeployabilityIndex] = None,
 215        **kwargs: t.Any,
 216    ) -> DF:
 217        """Renders the snapshot's model, executes it and returns a dataframe with the result.
 218
 219        Args:
 220            snapshot: Snapshot to evaluate.
 221            start: The start datetime to render.
 222            end: The end datetime to render.
 223            execution_time: The date/time time reference to use for execution time.
 224            snapshots: All upstream snapshots (by name) to use for expansion and mapping of physical locations.
 225            limit: The maximum number of rows to fetch.
 226            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 227            kwargs: Additional kwargs to pass to the renderer.
 228
 229        Returns:
 230            The result of the evaluation as a dataframe.
 231        """
 232        import pandas as pd
 233
 234        adapter = self.get_adapter(snapshot.model.gateway)
 235        render_kwargs = dict(
 236            start=start,
 237            end=end,
 238            execution_time=execution_time,
 239            snapshot=snapshot,
 240            runtime_stage=RuntimeStage.EVALUATING,
 241            **kwargs,
 242        )
 243        queries_or_dfs = self._render_snapshot_for_evaluation(
 244            snapshot,
 245            snapshots,
 246            deployability_index or DeployabilityIndex.all_deployable(),
 247            render_kwargs,
 248        )
 249        query_or_df = next(queries_or_dfs)
 250        if isinstance(query_or_df, pd.DataFrame):
 251            return query_or_df.head(limit)
 252        if not isinstance(query_or_df, exp.Expr):
 253            # We assume that if this branch is reached, `query_or_df` is a pyspark / snowpark / bigframe dataframe,
 254            # so we use `limit` instead of `head` to get back a dataframe instead of List[Row]
 255            # https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrame.head.html#pyspark.sql.DataFrame.head
 256            return query_or_df.limit(limit)
 257
 258        assert isinstance(query_or_df, exp.Query)
 259
 260        existing_limit = query_or_df.args.get("limit")
 261        if existing_limit:
 262            limit = min(limit, execute(exp.select(existing_limit.expression)).rows[0][0])
 263            assert limit is not None
 264
 265        return adapter._fetch_native_df(query_or_df.limit(limit))
 266
 267    def promote(
 268        self,
 269        target_snapshots: t.Iterable[Snapshot],
 270        environment_naming_info: EnvironmentNamingInfo,
 271        deployability_index: t.Optional[DeployabilityIndex] = None,
 272        start: t.Optional[TimeLike] = None,
 273        end: t.Optional[TimeLike] = None,
 274        execution_time: t.Optional[TimeLike] = None,
 275        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
 276        table_mapping: t.Optional[t.Dict[str, str]] = None,
 277        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
 278    ) -> None:
 279        """Promotes the given collection of snapshots in the target environment by replacing a corresponding
 280        view with a physical table associated with the given snapshot.
 281
 282        Args:
 283            target_snapshots: Snapshots to promote.
 284            environment_naming_info: Naming information for the target environment.
 285            deployability_index: Determines snapshots that are deployable in the context of this promotion.
 286            on_complete: A callback to call on each successfully promoted snapshot.
 287        """
 288
 289        tables_by_gateway: t.Dict[t.Union[str, None], t.List[exp.Table]] = defaultdict(list)
 290        for snapshot in target_snapshots:
 291            if snapshot.is_model and not snapshot.is_symbolic:
 292                gateway = (
 293                    snapshot.model_gateway if environment_naming_info.gateway_managed else None
 294                )
 295                adapter = self.get_adapter(gateway)
 296                table = snapshot.qualified_view_name.table_for_environment(
 297                    environment_naming_info, dialect=adapter.dialect
 298                )
 299                tables_by_gateway[gateway].append(table)
 300
 301        # A schema can be shared across multiple engines, so we need to group by gateway
 302        for gateway, tables in tables_by_gateway.items():
 303            if environment_naming_info.suffix_target.is_catalog:
 304                self._create_catalogs(tables=tables, gateway=gateway)
 305
 306        gateway_table_pairs = [
 307            (gateway, table) for gateway, tables in tables_by_gateway.items() for table in tables
 308        ]
 309        self._create_schemas(gateway_table_pairs=gateway_table_pairs)
 310
 311        # Fetch the view data objects for the promoted snapshots to get them cached
 312        self._get_virtual_data_objects(target_snapshots, environment_naming_info)
 313
 314        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 315        with self.concurrent_context():
 316            concurrent_apply_to_snapshots(
 317                target_snapshots,
 318                lambda s: self._promote_snapshot(
 319                    s,
 320                    start=start,
 321                    end=end,
 322                    execution_time=execution_time,
 323                    snapshots=snapshots,
 324                    table_mapping=table_mapping,
 325                    environment_naming_info=environment_naming_info,
 326                    deployability_index=deployability_index,  # type: ignore
 327                    on_complete=on_complete,
 328                ),
 329                self.ddl_concurrent_tasks,
 330            )
 331
 332    def demote(
 333        self,
 334        target_snapshots: t.Iterable[Snapshot],
 335        environment_naming_info: EnvironmentNamingInfo,
 336        table_mapping: t.Optional[t.Dict[str, str]] = None,
 337        deployability_index: t.Optional[DeployabilityIndex] = None,
 338        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
 339    ) -> None:
 340        """Demotes the given collection of snapshots in the target environment by removing its view.
 341
 342        Args:
 343            target_snapshots: Snapshots to demote.
 344            environment_naming_info: Naming info for the target environment.
 345            on_complete: A callback to call on each successfully demoted snapshot.
 346        """
 347        with self.concurrent_context():
 348            concurrent_apply_to_snapshots(
 349                target_snapshots,
 350                lambda s: self._demote_snapshot(
 351                    s,
 352                    environment_naming_info,
 353                    deployability_index=deployability_index,
 354                    on_complete=on_complete,
 355                    table_mapping=table_mapping,
 356                ),
 357                self.ddl_concurrent_tasks,
 358            )
 359
 360    def create(
 361        self,
 362        target_snapshots: t.Iterable[Snapshot],
 363        snapshots: t.Dict[SnapshotId, Snapshot],
 364        deployability_index: t.Optional[DeployabilityIndex] = None,
 365        on_start: t.Optional[t.Callable] = None,
 366        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
 367        allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
 368        allow_additive_snapshots: t.Optional[t.Set[str]] = None,
 369    ) -> CompletionStatus:
 370        """Creates a physical snapshot schema and table for the given collection of snapshots.
 371
 372        Args:
 373            target_snapshots: Target snapshots.
 374            snapshots: Mapping of snapshot ID to snapshot.
 375            deployability_index: Determines snapshots that are deployable in the context of this creation.
 376            on_start: A callback to initialize the snapshot creation progress bar.
 377            on_complete: A callback to call on each successfully created snapshot.
 378            allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
 379            allow_additive_snapshots: Set of snapshots that are allowed to have additive schema changes.
 380
 381        Returns:
 382            CompletionStatus: The status of the creation operation (success, failure, nothing to do).
 383        """
 384        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 385
 386        snapshots_to_create = self.get_snapshots_to_create(target_snapshots, deployability_index)
 387        if not snapshots_to_create:
 388            return CompletionStatus.NOTHING_TO_DO
 389        if on_start:
 390            on_start(snapshots_to_create)
 391
 392        self._create_snapshots(
 393            snapshots_to_create=snapshots_to_create,
 394            snapshots={s.name: s for s in snapshots.values()},
 395            deployability_index=deployability_index,
 396            on_complete=on_complete,
 397            allow_destructive_snapshots=allow_destructive_snapshots or set(),
 398            allow_additive_snapshots=allow_additive_snapshots or set(),
 399        )
 400        return CompletionStatus.SUCCESS
 401
 402    def create_physical_schemas(
 403        self, snapshots: t.Iterable[Snapshot], deployability_index: DeployabilityIndex
 404    ) -> None:
 405        """Creates the physical schemas for the given snapshots.
 406
 407        Args:
 408            snapshots: Snapshots to create physical schemas for.
 409            deployability_index: Determines snapshots that are deployable in the context of this creation.
 410        """
 411        tables_by_gateway: t.Dict[t.Optional[str], t.List[str]] = defaultdict(list)
 412        for snapshot in snapshots:
 413            if snapshot.is_model and not snapshot.is_symbolic:
 414                tables_by_gateway[snapshot.model_gateway].append(
 415                    snapshot.table_name(is_deployable=deployability_index.is_deployable(snapshot))
 416                )
 417
 418        gateway_table_pairs = [
 419            (gateway, table) for gateway, tables in tables_by_gateway.items() for table in tables
 420        ]
 421        self._create_schemas(gateway_table_pairs=gateway_table_pairs)
 422
 423    def get_snapshots_to_create(
 424        self, target_snapshots: t.Iterable[Snapshot], deployability_index: DeployabilityIndex
 425    ) -> t.List[Snapshot]:
 426        """Returns a list of snapshots that need to have their physical tables created.
 427
 428        Args:
 429            target_snapshots: Target snapshots.
 430            deployability_index: Determines snapshots that are deployable / representative in the context of this creation.
 431        """
 432        existing_data_objects = self._get_physical_data_objects(
 433            target_snapshots, deployability_index
 434        )
 435        snapshots_to_create = []
 436        for snapshot in target_snapshots:
 437            if not snapshot.is_model or snapshot.is_symbolic:
 438                continue
 439            if snapshot.snapshot_id not in existing_data_objects or (
 440                snapshot.is_seed and not snapshot.intervals
 441            ):
 442                snapshots_to_create.append(snapshot)
 443
 444        return snapshots_to_create
 445
 446    def _create_snapshots(
 447        self,
 448        snapshots_to_create: t.Iterable[Snapshot],
 449        snapshots: t.Dict[str, Snapshot],
 450        deployability_index: DeployabilityIndex,
 451        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
 452        allow_destructive_snapshots: t.Set[str],
 453        allow_additive_snapshots: t.Set[str],
 454    ) -> None:
 455        """Internal method to create tables in parallel."""
 456        with self.concurrent_context():
 457            errors, skipped = concurrent_apply_to_snapshots(
 458                snapshots_to_create,
 459                lambda s: self.create_snapshot(
 460                    s,
 461                    snapshots=snapshots,
 462                    deployability_index=deployability_index,
 463                    allow_destructive_snapshots=allow_destructive_snapshots,
 464                    allow_additive_snapshots=allow_additive_snapshots,
 465                    on_complete=on_complete,
 466                ),
 467                self.ddl_concurrent_tasks,
 468                raise_on_error=False,
 469            )
 470            if errors:
 471                raise SnapshotCreationFailedError(errors, skipped)
 472
 473    def migrate(
 474        self,
 475        target_snapshots: t.Iterable[Snapshot],
 476        snapshots: t.Dict[SnapshotId, Snapshot],
 477        allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
 478        allow_additive_snapshots: t.Optional[t.Set[str]] = None,
 479        deployability_index: t.Optional[DeployabilityIndex] = None,
 480    ) -> None:
 481        """Alters a physical snapshot table to match its snapshot's schema for the given collection of snapshots.
 482
 483        Args:
 484            target_snapshots: Target snapshots.
 485            snapshots: Mapping of snapshot ID to snapshot.
 486            allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
 487            allow_additive_snapshots: Set of snapshots that are allowed to have additive schema changes.
 488            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 489        """
 490        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 491        target_data_objects = self._get_physical_data_objects(target_snapshots, deployability_index)
 492        if not target_data_objects:
 493            return
 494
 495        if not snapshots:
 496            snapshots = {s.snapshot_id: s for s in target_snapshots}
 497
 498        allow_destructive_snapshots = allow_destructive_snapshots or set()
 499        allow_additive_snapshots = allow_additive_snapshots or set()
 500        snapshots_by_name = {s.name: s for s in snapshots.values()}
 501        with self.concurrent_context():
 502            # Only migrate snapshots for which there's an existing data object
 503            concurrent_apply_to_snapshots(
 504                target_snapshots,
 505                lambda s: self._migrate_snapshot(
 506                    s,
 507                    snapshots_by_name,
 508                    target_data_objects.get(s.snapshot_id),
 509                    allow_destructive_snapshots,
 510                    allow_additive_snapshots,
 511                    self.get_adapter(s.model_gateway),
 512                    deployability_index,
 513                ),
 514                self.ddl_concurrent_tasks,
 515            )
 516
 517    def cleanup(
 518        self,
 519        target_snapshots: t.Iterable[SnapshotTableCleanupTask],
 520        on_complete: t.Optional[t.Callable[[str], None]] = None,
 521    ) -> None:
 522        """Cleans up the given snapshots by removing its table
 523
 524        Args:
 525            target_snapshots: Snapshots to cleanup.
 526            on_complete: A callback to call on each successfully deleted database object.
 527        """
 528        target_snapshots = [
 529            t for t in target_snapshots if t.snapshot.is_model and not t.snapshot.is_symbolic
 530        ]
 531        available_gateways = set(self.adapters.keys())
 532        skipped = []
 533        filtered_targets = []
 534        for t in target_snapshots:
 535            gw = t.snapshot.model_gateway
 536            if gw and gw not in available_gateways:
 537                skipped.append((t.snapshot.snapshot_id, gw))
 538            else:
 539                filtered_targets.append(t)
 540        if skipped:
 541            logger.warning(
 542                "Skipping cleanup of %d snapshot(s) with unavailable gateway(s): %s",
 543                len(skipped),
 544                ", ".join(f"{sid} (gateway={gw})" for sid, gw in skipped),
 545            )
 546        snapshots_to_dev_table_only = {
 547            t.snapshot.snapshot_id: t.dev_table_only for t in filtered_targets
 548        }
 549        with self.concurrent_context():
 550            errors, _ = concurrent_apply_to_snapshots(
 551                [t.snapshot for t in filtered_targets],
 552                lambda s: self._cleanup_snapshot(
 553                    s,
 554                    snapshots_to_dev_table_only[s.snapshot_id],
 555                    self.get_adapter(s.model_gateway),
 556                    on_complete,
 557                ),
 558                self.ddl_concurrent_tasks,
 559                reverse_order=True,
 560                raise_on_error=False,
 561            )
 562        if errors:
 563            errored_snapshots = "\n".join(f"  {e.node.name}: {e.__cause__}" for e in errors)
 564            raise SQLMeshError(f"\n{errored_snapshots}")
 565
 566    def audit(
 567        self,
 568        snapshot: Snapshot,
 569        *,
 570        snapshots: t.Dict[str, Snapshot],
 571        start: t.Optional[TimeLike] = None,
 572        end: t.Optional[TimeLike] = None,
 573        execution_time: t.Optional[TimeLike] = None,
 574        deployability_index: t.Optional[DeployabilityIndex] = None,
 575        wap_id: t.Optional[str] = None,
 576        **kwargs: t.Any,
 577    ) -> t.List[AuditResult]:
 578        """Execute a snapshot's node's audit queries.
 579
 580        Args:
 581            snapshot: Snapshot to evaluate.
 582            snapshots: All upstream snapshots (by name) to use for expansion and mapping of physical locations.
 583            start: The start datetime to audit. Defaults to epoch start.
 584            end: The end datetime to audit. Defaults to epoch start.
 585            execution_time: The date/time time reference to use for execution time.
 586            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 587            wap_id: The WAP ID if applicable, None otherwise.
 588            kwargs: Additional kwargs to pass to the renderer.
 589        """
 590        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 591        adapter = self.get_adapter(snapshot.model_gateway)
 592
 593        if not snapshot.version:
 594            raise ConfigError(
 595                f"Cannot audit '{snapshot.name}' because it has not been versioned yet. Apply a plan first."
 596            )
 597
 598        if wap_id is not None:
 599            deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 600            original_table_name = snapshot.table_name(
 601                is_deployable=deployability_index.is_deployable(snapshot)
 602            )
 603            wap_table_name = adapter.wap_table_name(original_table_name, wap_id)
 604            logger.info(
 605                "Auditing WAP table '%s', snapshot %s",
 606                wap_table_name,
 607                snapshot.snapshot_id,
 608            )
 609
 610            table_mapping = kwargs.get("table_mapping") or {}
 611            table_mapping[snapshot.name] = wap_table_name
 612            kwargs["table_mapping"] = table_mapping
 613            kwargs["this_model"] = exp.to_table(wap_table_name, dialect=adapter.dialect)
 614
 615        results = []
 616
 617        audits_with_args = snapshot.node.audits_with_args
 618
 619        force_non_blocking = False
 620
 621        if audits_with_args:
 622            logger.info("Auditing snapshot %s", snapshot.snapshot_id)
 623
 624            if not deployability_index.is_deployable(snapshot) and not adapter.SUPPORTS_CLONING:
 625                # For dev preview tables that aren't based on clones of the production table, only a subset of the data is typically available
 626                # However, users still expect audits to run anwyay. Some audits (such as row count) are practically guaranteed to fail
 627                # when run on only a subset of data, so we switch all audits to non blocking and the user can decide if they still want to proceed
 628                force_non_blocking = True
 629
 630        for audit, audit_args in audits_with_args:
 631            if force_non_blocking:
 632                # remove any blocking indicator on the model itself
 633                audit_args.pop("blocking", None)
 634                # so that we can fall back to the audit's setting, which we override to blocking: False
 635                audit = audit.model_copy(update={"blocking": False})
 636
 637            results.append(
 638                self._audit(
 639                    audit=audit,
 640                    audit_args=audit_args,
 641                    snapshot=snapshot,
 642                    snapshots=snapshots,
 643                    start=start,
 644                    end=end,
 645                    execution_time=execution_time,
 646                    deployability_index=deployability_index,
 647                    **kwargs,
 648                )
 649            )
 650
 651        if wap_id is not None:
 652            logger.info(
 653                "Publishing evaluation results for snapshot %s, WAP ID '%s'",
 654                snapshot.snapshot_id,
 655                wap_id,
 656            )
 657            self.wap_publish_snapshot(snapshot, wap_id, deployability_index)
 658
 659        return results
 660
 661    @contextmanager
 662    def concurrent_context(self) -> t.Iterator[None]:
 663        try:
 664            yield
 665        finally:
 666            self.recycle()
 667
 668    def recycle(self) -> None:
 669        """Closes all open connections and releases all allocated resources associated with any thread
 670        except the calling one."""
 671        try:
 672            for adapter in self.adapters.values():
 673                adapter.recycle()
 674
 675        except Exception:
 676            logger.exception("Failed to recycle Snapshot Evaluator")
 677
 678    def close(self) -> None:
 679        """Closes all open connections and releases all allocated resources."""
 680        try:
 681            for adapter in self.adapters.values():
 682                adapter.close()
 683        except Exception:
 684            logger.exception("Failed to close Snapshot Evaluator")
 685
 686    def set_correlation_id(self, correlation_id: CorrelationId) -> SnapshotEvaluator:
 687        return SnapshotEvaluator(
 688            {
 689                gateway: adapter.with_settings(correlation_id=correlation_id)
 690                for gateway, adapter in self.adapters.items()
 691            },
 692            self.ddl_concurrent_tasks,
 693            self.selected_gateway,
 694        )
 695
 696    def _evaluate_snapshot(
 697        self,
 698        start: TimeLike,
 699        end: TimeLike,
 700        execution_time: TimeLike,
 701        snapshot: Snapshot,
 702        snapshots: t.Dict[str, Snapshot],
 703        allow_destructive_snapshots: t.Set[str],
 704        allow_additive_snapshots: t.Set[str],
 705        deployability_index: t.Optional[DeployabilityIndex],
 706        batch_index: int,
 707        target_table_exists: t.Optional[bool],
 708        **kwargs: t.Any,
 709    ) -> t.Optional[str]:
 710        """Renders the snapshot's model and executes it. The return value depends on whether the limit was specified.
 711
 712        Args:
 713            snapshot: Snapshot to evaluate.
 714            start: The start datetime to render.
 715            end: The end datetime to render.
 716            execution_time: The date/time time reference to use for execution time.
 717            snapshots: All upstream snapshots to use for expansion and mapping of physical locations.
 718            allow_destructive_snapshots: Snapshots for which destructive schema changes are allowed.
 719            allow_additive_snapshots: Snapshots for which additive schema changes are allowed.
 720            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 721            batch_index: If the snapshot is part of a batch of related snapshots; which index in the batch is it
 722            target_table_exists: Whether the target table exists. If None, the table will be checked for existence.
 723            kwargs: Additional kwargs to pass to the renderer.
 724        """
 725        if not snapshot.is_model:
 726            return None
 727
 728        model = snapshot.model
 729
 730        logger.info("Evaluating snapshot %s", snapshot.snapshot_id)
 731
 732        adapter = self.get_adapter(model.gateway)
 733        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 734        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
 735        target_table_name = snapshot.table_name(is_deployable=is_snapshot_deployable)
 736        # https://github.com/SQLMesh/sqlmesh/issues/2609
 737        # If there are no existing intervals yet; only consider this a first insert for the first snapshot in the batch
 738        if target_table_exists is None:
 739            target_table_exists = adapter.table_exists(target_table_name)
 740        is_first_insert = (
 741            not _intervals(snapshot, deployability_index) or not target_table_exists
 742        ) and batch_index == 0
 743
 744        # Use the 'creating' stage if the table doesn't exist yet to preserve backwards compatibility with existing projects
 745        # that depend on a separate physical table creation stage.
 746        runtime_stage = RuntimeStage.EVALUATING if target_table_exists else RuntimeStage.CREATING
 747        common_render_kwargs = dict(
 748            start=start,
 749            end=end,
 750            execution_time=execution_time,
 751            snapshot=snapshot,
 752            runtime_stage=runtime_stage,
 753            **kwargs,
 754        )
 755        create_render_kwargs = dict(
 756            engine_adapter=adapter,
 757            snapshots=snapshots,
 758            deployability_index=deployability_index,
 759            **common_render_kwargs,
 760        )
 761        create_render_kwargs["runtime_stage"] = RuntimeStage.CREATING
 762        render_statements_kwargs = dict(
 763            engine_adapter=adapter,
 764            snapshots=snapshots,
 765            deployability_index=deployability_index,
 766            **common_render_kwargs,
 767        )
 768        rendered_physical_properties = snapshot.model.render_physical_properties(
 769            **render_statements_kwargs
 770        )
 771
 772        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
 773        evaluation_strategy.run_pre_statements(
 774            snapshot=snapshot,
 775            render_kwargs={**render_statements_kwargs, "inside_transaction": False},
 776        )
 777
 778        with (
 779            adapter.transaction(),
 780            adapter.session(snapshot.model.render_session_properties(**render_statements_kwargs)),
 781        ):
 782            evaluation_strategy.run_pre_statements(
 783                snapshot=snapshot,
 784                render_kwargs={**render_statements_kwargs, "inside_transaction": True},
 785            )
 786
 787            if not target_table_exists or (model.is_seed and not snapshot.intervals):
 788                # Only create the empty table if the columns were provided explicitly by the user
 789                should_create_empty_table = (
 790                    model.kind.is_materialized
 791                    and model.columns_to_types_
 792                    and columns_to_types_all_known(model.columns_to_types_)
 793                )
 794                if not should_create_empty_table:
 795                    # Or if the model is self-referential and its query is fully annotated with types
 796                    should_create_empty_table = model.depends_on_self and model.annotated
 797                if self._can_clone(snapshot, deployability_index):
 798                    self._clone_snapshot_in_dev(
 799                        snapshot=snapshot,
 800                        snapshots=snapshots,
 801                        deployability_index=deployability_index,
 802                        render_kwargs=create_render_kwargs,
 803                        rendered_physical_properties=rendered_physical_properties.copy(),
 804                        allow_destructive_snapshots=allow_destructive_snapshots,
 805                        allow_additive_snapshots=allow_additive_snapshots,
 806                    )
 807                    runtime_stage = RuntimeStage.EVALUATING
 808                    target_table_exists = True
 809                elif should_create_empty_table or model.is_seed or model.kind.is_scd_type_2:
 810                    self._execute_create(
 811                        snapshot=snapshot,
 812                        table_name=target_table_name,
 813                        is_table_deployable=is_snapshot_deployable,
 814                        deployability_index=deployability_index,
 815                        create_render_kwargs=create_render_kwargs,
 816                        rendered_physical_properties=rendered_physical_properties.copy(),
 817                        dry_run=False,
 818                        run_pre_post_statements=False,
 819                    )
 820                    runtime_stage = RuntimeStage.EVALUATING
 821                    target_table_exists = True
 822
 823            evaluate_render_kwargs = {
 824                **common_render_kwargs,
 825                "runtime_stage": runtime_stage,
 826                "snapshot_table_exists": target_table_exists,
 827            }
 828
 829            wap_id: t.Optional[str] = None
 830            if (
 831                snapshot.is_materialized
 832                and target_table_exists
 833                and adapter.wap_enabled
 834                and (model.wap_supported or adapter.wap_supported(target_table_name))
 835            ):
 836                wap_id = random_id()[0:8]
 837                logger.info("Using WAP ID '%s' for snapshot %s", wap_id, snapshot.snapshot_id)
 838                target_table_name = adapter.wap_prepare(target_table_name, wap_id)
 839
 840            self._render_and_insert_snapshot(
 841                start=start,
 842                end=end,
 843                execution_time=execution_time,
 844                snapshot=snapshot,
 845                snapshots=snapshots,
 846                render_kwargs=evaluate_render_kwargs,
 847                create_render_kwargs=create_render_kwargs,
 848                rendered_physical_properties=rendered_physical_properties,
 849                deployability_index=deployability_index,
 850                target_table_name=target_table_name,
 851                is_first_insert=is_first_insert,
 852                batch_index=batch_index,
 853            )
 854
 855            evaluation_strategy.run_post_statements(
 856                snapshot=snapshot,
 857                render_kwargs={**render_statements_kwargs, "inside_transaction": True},
 858            )
 859
 860        evaluation_strategy.run_post_statements(
 861            snapshot=snapshot,
 862            render_kwargs={**render_statements_kwargs, "inside_transaction": False},
 863        )
 864
 865        return wap_id
 866
 867    def create_snapshot(
 868        self,
 869        snapshot: Snapshot,
 870        snapshots: t.Dict[str, Snapshot],
 871        deployability_index: DeployabilityIndex,
 872        allow_destructive_snapshots: t.Set[str],
 873        allow_additive_snapshots: t.Set[str],
 874        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
 875    ) -> None:
 876        """Creates a physical table for the given snapshot.
 877
 878        Args:
 879            snapshot: Snapshot to create.
 880            snapshots: All upstream snapshots to use for expansion and mapping of physical locations.
 881            deployability_index: Determines snapshots that are deployable in the context of this creation.
 882            on_complete: A callback to call on each successfully created database object.
 883            allow_destructive_snapshots: Snapshots for which destructive schema changes are allowed.
 884            allow_additive_snapshots: Snapshots for which additive schema changes are allowed.
 885        """
 886        if not snapshot.is_model:
 887            return
 888
 889        logger.info("Creating a physical table for snapshot %s", snapshot.snapshot_id)
 890
 891        adapter = self.get_adapter(snapshot.model.gateway)
 892        create_render_kwargs: t.Dict[str, t.Any] = dict(
 893            engine_adapter=adapter,
 894            snapshots=snapshots,
 895            runtime_stage=RuntimeStage.CREATING,
 896            deployability_index=deployability_index,
 897        )
 898
 899        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
 900        evaluation_strategy.run_pre_statements(
 901            snapshot=snapshot, render_kwargs={**create_render_kwargs, "inside_transaction": False}
 902        )
 903
 904        with (
 905            adapter.transaction(),
 906            adapter.session(snapshot.model.render_session_properties(**create_render_kwargs)),
 907        ):
 908            rendered_physical_properties = snapshot.model.render_physical_properties(
 909                **create_render_kwargs
 910            )
 911
 912            if self._can_clone(snapshot, deployability_index):
 913                self._clone_snapshot_in_dev(
 914                    snapshot=snapshot,
 915                    snapshots=snapshots,
 916                    deployability_index=deployability_index,
 917                    render_kwargs=create_render_kwargs,
 918                    rendered_physical_properties=rendered_physical_properties,
 919                    allow_destructive_snapshots=allow_destructive_snapshots,
 920                    allow_additive_snapshots=allow_additive_snapshots,
 921                    run_pre_post_statements=True,
 922                )
 923            else:
 924                is_table_deployable = deployability_index.is_deployable(snapshot)
 925                self._execute_create(
 926                    snapshot=snapshot,
 927                    table_name=snapshot.table_name(is_deployable=is_table_deployable),
 928                    is_table_deployable=is_table_deployable,
 929                    deployability_index=deployability_index,
 930                    create_render_kwargs=create_render_kwargs,
 931                    rendered_physical_properties=rendered_physical_properties,
 932                    dry_run=True,
 933                )
 934
 935        evaluation_strategy.run_post_statements(
 936            snapshot=snapshot, render_kwargs={**create_render_kwargs, "inside_transaction": False}
 937        )
 938
 939        if on_complete is not None:
 940            on_complete(snapshot)
 941
 942    def wap_publish_snapshot(
 943        self,
 944        snapshot: Snapshot,
 945        wap_id: str,
 946        deployability_index: t.Optional[DeployabilityIndex],
 947    ) -> None:
 948        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
 949        table_name = snapshot.table_name(is_deployable=deployability_index.is_deployable(snapshot))
 950        adapter = self.get_adapter(snapshot.model_gateway)
 951        adapter.wap_publish(table_name, wap_id)
 952
 953    def _render_and_insert_snapshot(
 954        self,
 955        start: TimeLike,
 956        end: TimeLike,
 957        execution_time: TimeLike,
 958        snapshot: Snapshot,
 959        snapshots: t.Dict[str, Snapshot],
 960        render_kwargs: t.Dict[str, t.Any],
 961        create_render_kwargs: t.Dict[str, t.Any],
 962        rendered_physical_properties: t.Dict[str, exp.Expr],
 963        deployability_index: DeployabilityIndex,
 964        target_table_name: str,
 965        is_first_insert: bool,
 966        batch_index: int,
 967    ) -> None:
 968        if not snapshot.is_model or snapshot.is_seed:
 969            return
 970
 971        logger.info("Inserting data for snapshot %s", snapshot.snapshot_id)
 972
 973        model = snapshot.model
 974        adapter = self.get_adapter(model.gateway)
 975        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
 976        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
 977
 978        queries_or_dfs = self._render_snapshot_for_evaluation(
 979            snapshot,
 980            snapshots,
 981            deployability_index,
 982            render_kwargs,
 983        )
 984
 985        def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
 986            if index > 0:
 987                evaluation_strategy.append(
 988                    table_name=target_table_name,
 989                    query_or_df=query_or_df,
 990                    model=snapshot.model,
 991                    snapshot=snapshot,
 992                    snapshots=snapshots,
 993                    deployability_index=deployability_index,
 994                    batch_index=batch_index,
 995                    start=start,
 996                    end=end,
 997                    execution_time=execution_time,
 998                    physical_properties=rendered_physical_properties,
 999                    render_kwargs=create_render_kwargs,
1000                    is_snapshot_deployable=is_snapshot_deployable,
1001                )
1002            else:
1003                logger.info(
1004                    "Inserting batch (%s, %s) into %s'",
1005                    time_like_to_str(start),
1006                    time_like_to_str(end),
1007                    target_table_name,
1008                )
1009                evaluation_strategy.insert(
1010                    table_name=target_table_name,
1011                    query_or_df=query_or_df,
1012                    is_first_insert=is_first_insert,
1013                    model=snapshot.model,
1014                    snapshot=snapshot,
1015                    snapshots=snapshots,
1016                    deployability_index=deployability_index,
1017                    batch_index=batch_index,
1018                    start=start,
1019                    end=end,
1020                    execution_time=execution_time,
1021                    physical_properties=rendered_physical_properties,
1022                    render_kwargs=create_render_kwargs,
1023                    is_snapshot_deployable=is_snapshot_deployable,
1024                )
1025
1026        # DataFrames, unlike SQL expressions, can provide partial results by yielding dataframes. As a result,
1027        # if the engine supports INSERT OVERWRITE or REPLACE WHERE and the snapshot is incremental by time range, we risk
1028        # having a partial result since each dataframe write can re-truncate partitions. To avoid this, we
1029        # union all the dataframes together before writing. For pandas this could result in OOM and a potential
1030        # workaround for that would be to serialize pandas to disk and then read it back with Spark.
1031        # Note: We assume that if multiple things are yielded from `queries_or_dfs` that they are dataframes
1032        # and not SQL expressions.
1033        if (
1034            adapter.INSERT_OVERWRITE_STRATEGY
1035            in (
1036                InsertOverwriteStrategy.INSERT_OVERWRITE,
1037                InsertOverwriteStrategy.REPLACE_WHERE,
1038            )
1039            and snapshot.is_incremental_by_time_range
1040        ):
1041            import pandas as pd
1042
1043            try:
1044                first_query_or_df = next(queries_or_dfs)
1045            except StopIteration:
1046                return
1047
1048            query_or_df = reduce(
1049                lambda a, b: (
1050                    pd.concat([a, b], ignore_index=True)  # type: ignore
1051                    if isinstance(a, pd.DataFrame)
1052                    else a.union_all(b)  # type: ignore
1053                ),  # type: ignore
1054                queries_or_dfs,
1055                first_query_or_df,
1056            )
1057            apply(query_or_df, index=0)
1058        else:
1059            for index, query_or_df in enumerate(queries_or_dfs):
1060                apply(query_or_df, index)
1061
1062    def _render_snapshot_for_evaluation(
1063        self,
1064        snapshot: Snapshot,
1065        snapshots: t.Dict[str, Snapshot],
1066        deployability_index: DeployabilityIndex,
1067        render_kwargs: t.Dict[str, t.Any],
1068    ) -> t.Iterator[QueryOrDF]:
1069        from sqlmesh.core.context import ExecutionContext
1070
1071        model = snapshot.model
1072        adapter = self.get_adapter(model.gateway)
1073
1074        return model.render(
1075            context=ExecutionContext(
1076                adapter,
1077                snapshots,
1078                deployability_index,
1079                default_dialect=model.dialect,
1080                default_catalog=model.default_catalog,
1081            ),
1082            **render_kwargs,
1083        )
1084
1085    def _clone_snapshot_in_dev(
1086        self,
1087        snapshot: Snapshot,
1088        snapshots: t.Dict[str, Snapshot],
1089        deployability_index: DeployabilityIndex,
1090        render_kwargs: t.Dict[str, t.Any],
1091        rendered_physical_properties: t.Dict[str, exp.Expr],
1092        allow_destructive_snapshots: t.Set[str],
1093        allow_additive_snapshots: t.Set[str],
1094        run_pre_post_statements: bool = False,
1095    ) -> None:
1096        adapter = self.get_adapter(snapshot.model.gateway)
1097
1098        target_table_name = snapshot.table_name(is_deployable=False)
1099        source_table_name = snapshot.table_name()
1100
1101        try:
1102            logger.info(f"Cloning table '{source_table_name}' into '{target_table_name}'")
1103            adapter.clone_table(
1104                target_table_name,
1105                snapshot.table_name(),
1106                rendered_physical_properties=rendered_physical_properties,
1107            )
1108            self._migrate_target_table(
1109                target_table_name=target_table_name,
1110                snapshot=snapshot,
1111                snapshots=snapshots,
1112                deployability_index=deployability_index,
1113                render_kwargs=render_kwargs,
1114                rendered_physical_properties=rendered_physical_properties,
1115                allow_destructive_snapshots=allow_destructive_snapshots,
1116                allow_additive_snapshots=allow_additive_snapshots,
1117                run_pre_post_statements=run_pre_post_statements,
1118            )
1119
1120        except Exception:
1121            adapter.drop_table(target_table_name)
1122            raise
1123
1124    def _migrate_snapshot(
1125        self,
1126        snapshot: Snapshot,
1127        snapshots: t.Dict[str, Snapshot],
1128        target_data_object: t.Optional[DataObject],
1129        allow_destructive_snapshots: t.Set[str],
1130        allow_additive_snapshots: t.Set[str],
1131        adapter: EngineAdapter,
1132        deployability_index: DeployabilityIndex,
1133    ) -> None:
1134        if not snapshot.is_model or snapshot.is_symbolic:
1135            return
1136
1137        deployability_index = DeployabilityIndex.all_deployable()
1138        render_kwargs: t.Dict[str, t.Any] = dict(
1139            engine_adapter=adapter,
1140            snapshots=snapshots,
1141            runtime_stage=RuntimeStage.CREATING,
1142            deployability_index=deployability_index,
1143        )
1144        target_table_name = snapshot.table_name()
1145
1146        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
1147        evaluation_strategy.run_pre_statements(
1148            snapshot=snapshot, render_kwargs={**render_kwargs, "inside_transaction": False}
1149        )
1150
1151        with (
1152            adapter.transaction(),
1153            adapter.session(snapshot.model.render_session_properties(**render_kwargs)),
1154        ):
1155            table_exists = target_data_object is not None
1156            if adapter.drop_data_object_on_type_mismatch(
1157                target_data_object, _snapshot_to_data_object_type(snapshot)
1158            ):
1159                table_exists = False
1160
1161            rendered_physical_properties = snapshot.model.render_physical_properties(
1162                **render_kwargs
1163            )
1164
1165            if table_exists:
1166                self._migrate_target_table(
1167                    target_table_name=target_table_name,
1168                    snapshot=snapshot,
1169                    snapshots=snapshots,
1170                    deployability_index=deployability_index,
1171                    render_kwargs=render_kwargs,
1172                    rendered_physical_properties=rendered_physical_properties,
1173                    allow_destructive_snapshots=allow_destructive_snapshots,
1174                    allow_additive_snapshots=allow_additive_snapshots,
1175                    run_pre_post_statements=True,
1176                )
1177            else:
1178                self._execute_create(
1179                    snapshot=snapshot,
1180                    table_name=snapshot.table_name(is_deployable=True),
1181                    is_table_deployable=True,
1182                    deployability_index=deployability_index,
1183                    create_render_kwargs=render_kwargs,
1184                    rendered_physical_properties=rendered_physical_properties,
1185                    dry_run=True,
1186                )
1187
1188        evaluation_strategy.run_post_statements(
1189            snapshot=snapshot, render_kwargs={**render_kwargs, "inside_transaction": False}
1190        )
1191
1192    # Retry in case when the table is migrated concurrently from another plan application
1193    @retry(
1194        reraise=True,
1195        stop=stop_after_attempt(5),
1196        wait=wait_exponential(min=1, max=16),
1197        retry=retry_if_not_exception_type(
1198            (DestructiveChangeError, AdditiveChangeError, MigrationNotSupportedError)
1199        ),
1200    )
1201    def _migrate_target_table(
1202        self,
1203        target_table_name: str,
1204        snapshot: Snapshot,
1205        snapshots: t.Dict[str, Snapshot],
1206        deployability_index: DeployabilityIndex,
1207        render_kwargs: t.Dict[str, t.Any],
1208        rendered_physical_properties: t.Dict[str, exp.Expr],
1209        allow_destructive_snapshots: t.Set[str],
1210        allow_additive_snapshots: t.Set[str],
1211        run_pre_post_statements: bool = False,
1212    ) -> None:
1213        adapter = self.get_adapter(snapshot.model.gateway)
1214
1215        tmp_table = exp.to_table(target_table_name)
1216        tmp_table.this.set("this", f"{tmp_table.name}_schema_tmp")
1217        tmp_table_name = tmp_table.sql()
1218
1219        if snapshot.is_materialized:
1220            self._execute_create(
1221                snapshot=snapshot,
1222                table_name=tmp_table_name,
1223                is_table_deployable=False,
1224                deployability_index=deployability_index,
1225                create_render_kwargs=render_kwargs,
1226                rendered_physical_properties=rendered_physical_properties,
1227                dry_run=False,
1228                run_pre_post_statements=run_pre_post_statements,
1229                skip_grants=True,  # skip grants for tmp table
1230            )
1231        try:
1232            evaluation_strategy = _evaluation_strategy(snapshot, adapter)
1233            logger.info(
1234                "Migrating table schema from '%s' to '%s'",
1235                tmp_table_name,
1236                target_table_name,
1237            )
1238            evaluation_strategy.migrate(
1239                target_table_name=target_table_name,
1240                source_table_name=tmp_table_name,
1241                snapshot=snapshot,
1242                snapshots=snapshots,
1243                allow_destructive_snapshots=allow_destructive_snapshots,
1244                allow_additive_snapshots=allow_additive_snapshots,
1245                ignore_destructive=snapshot.model.on_destructive_change.is_ignore,
1246                ignore_additive=snapshot.model.on_additive_change.is_ignore,
1247                deployability_index=deployability_index,
1248            )
1249        finally:
1250            if snapshot.is_materialized:
1251                adapter.drop_table(tmp_table_name)
1252
1253    def _promote_snapshot(
1254        self,
1255        snapshot: Snapshot,
1256        environment_naming_info: EnvironmentNamingInfo,
1257        deployability_index: DeployabilityIndex,
1258        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
1259        start: t.Optional[TimeLike] = None,
1260        end: t.Optional[TimeLike] = None,
1261        execution_time: t.Optional[TimeLike] = None,
1262        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
1263        table_mapping: t.Optional[t.Dict[str, str]] = None,
1264    ) -> None:
1265        if not snapshot.is_model:
1266            return
1267
1268        adapter = (
1269            self.get_adapter(snapshot.model_gateway)
1270            if environment_naming_info.gateway_managed
1271            else self.adapter
1272        )
1273        table_name = snapshot.table_name(deployability_index.is_representative(snapshot))
1274        view_name = snapshot.qualified_view_name.for_environment(
1275            environment_naming_info, dialect=adapter.dialect
1276        )
1277        render_kwargs: t.Dict[str, t.Any] = dict(
1278            start=start,
1279            end=end,
1280            execution_time=execution_time,
1281            engine_adapter=adapter,
1282            deployability_index=deployability_index,
1283            table_mapping=table_mapping,
1284            runtime_stage=RuntimeStage.PROMOTING,
1285        )
1286
1287        with (
1288            adapter.transaction(),
1289            adapter.session(snapshot.model.render_session_properties(**render_kwargs)),
1290        ):
1291            _evaluation_strategy(snapshot, adapter).promote(
1292                table_name=table_name,
1293                view_name=view_name,
1294                model=snapshot.model,
1295                environment=environment_naming_info.name,
1296                snapshots=snapshots,
1297                snapshot=snapshot,
1298                **render_kwargs,
1299            )
1300
1301            snapshot_by_name = {s.name: s for s in (snapshots or {}).values()}
1302            render_kwargs["snapshots"] = snapshot_by_name
1303            adapter.execute(snapshot.model.render_on_virtual_update(**render_kwargs))
1304
1305        if on_complete is not None:
1306            on_complete(snapshot)
1307
1308    def _demote_snapshot(
1309        self,
1310        snapshot: Snapshot,
1311        environment_naming_info: EnvironmentNamingInfo,
1312        deployability_index: t.Optional[DeployabilityIndex],
1313        on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
1314        table_mapping: t.Optional[t.Dict[str, str]] = None,
1315    ) -> None:
1316        if not snapshot.is_model:
1317            return
1318
1319        adapter = (
1320            self.get_adapter(snapshot.model_gateway)
1321            if environment_naming_info.gateway_managed
1322            else self.adapter
1323        )
1324        view_name = snapshot.qualified_view_name.for_environment(
1325            environment_naming_info, dialect=adapter.dialect
1326        )
1327        with (
1328            adapter.transaction(),
1329            adapter.session(
1330                snapshot.model.render_session_properties(
1331                    engine_adapter=adapter,
1332                    deployability_index=deployability_index,
1333                    table_mapping=table_mapping,
1334                    runtime_stage=RuntimeStage.DEMOTING,
1335                )
1336            ),
1337        ):
1338            _evaluation_strategy(snapshot, adapter).demote(view_name)
1339
1340        if on_complete is not None:
1341            on_complete(snapshot)
1342
1343    def _cleanup_snapshot(
1344        self,
1345        snapshot: SnapshotInfoLike,
1346        dev_table_only: bool,
1347        adapter: EngineAdapter,
1348        on_complete: t.Optional[t.Callable[[str], None]],
1349    ) -> None:
1350        snapshot = snapshot.table_info
1351
1352        table_names = [(False, snapshot.table_name(is_deployable=False))]
1353        if not dev_table_only:
1354            table_names.append((True, snapshot.table_name(is_deployable=True)))
1355
1356        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
1357        for is_table_deployable, table_name in table_names:
1358            try:
1359                evaluation_strategy.delete(
1360                    table_name,
1361                    is_table_deployable=is_table_deployable,
1362                    physical_schema=snapshot.physical_schema,
1363                    # we need to set cascade=true or we will get a 'cant drop because other objects depend on it'-style
1364                    # error on engines that enforce referential integrity, such as Postgres
1365                    # this situation can happen when a snapshot expires but downstream view snapshots that reference it have not yet expired
1366                    cascade=True,
1367                )
1368            except Exception:
1369                # Use `get_data_object` to check if the table exists instead of `table_exists` since the former
1370                # is based on `INFORMATION_SCHEMA` and avoids touching the table directly.
1371                # This is important when the table name is malformed for some reason and running any statement
1372                # that touches the table would result in an error.
1373                if adapter.get_data_object(table_name) is not None:
1374                    raise
1375                logger.warning(
1376                    "Skipping cleanup of table '%s' because it does not exist", table_name
1377                )
1378
1379            if on_complete is not None:
1380                on_complete(table_name)
1381
1382    def _audit(
1383        self,
1384        audit: Audit,
1385        audit_args: t.Dict[t.Any, t.Any],
1386        snapshot: Snapshot,
1387        snapshots: t.Dict[str, Snapshot],
1388        start: t.Optional[TimeLike],
1389        end: t.Optional[TimeLike],
1390        execution_time: t.Optional[TimeLike],
1391        deployability_index: t.Optional[DeployabilityIndex],
1392        **kwargs: t.Any,
1393    ) -> AuditResult:
1394        if audit.skip:
1395            return AuditResult(
1396                audit=audit,
1397                audit_args=audit_args,
1398                model=snapshot.model_or_none,
1399                skipped=True,
1400            )
1401
1402        # Model's "blocking" argument takes precedence over the audit's default setting
1403        blocking = audit_args.pop("blocking", None)
1404        blocking = blocking == exp.true() if blocking else audit.blocking
1405
1406        adapter = self.get_adapter(snapshot.model_gateway)
1407
1408        kwargs = {
1409            "start": start,
1410            "end": end,
1411            "execution_time": execution_time,
1412            "snapshots": snapshots,
1413            "deployability_index": deployability_index,
1414            "engine_adapter": adapter,
1415            "runtime_stage": RuntimeStage.AUDITING,
1416            **audit_args,
1417            **kwargs,
1418        }
1419
1420        if snapshot.is_model:
1421            query = snapshot.model.render_audit_query(audit, **kwargs)
1422        elif isinstance(audit, StandaloneAudit):
1423            query = audit.render_audit_query(**kwargs)
1424        else:
1425            raise SQLMeshError("Expected model or standalone audit. {snapshot}: {audit}")
1426
1427        count, *_ = adapter.fetchone(
1428            select("COUNT(*)").from_(query.subquery("audit")),
1429            quote_identifiers=True,
1430        )  # type: ignore
1431
1432        return AuditResult(
1433            audit=audit,
1434            audit_args=audit_args,
1435            model=snapshot.model_or_none,
1436            count=count,
1437            query=query,
1438            blocking=blocking,
1439        )
1440
1441    def _create_catalogs(
1442        self,
1443        tables: t.Iterable[t.Union[exp.Table, str]],
1444        gateway: t.Optional[str] = None,
1445    ) -> None:
1446        # attempt to create catalogs for the virtual layer if possible
1447        adapter = self.get_adapter(gateway)
1448        if adapter.SUPPORTS_CREATE_DROP_CATALOG:
1449            unique_catalogs = {t.catalog for t in [exp.to_table(maybe_t) for maybe_t in tables]}
1450            for catalog_name in unique_catalogs:
1451                adapter.create_catalog(catalog_name)
1452
1453    def _create_schemas(
1454        self,
1455        gateway_table_pairs: t.Iterable[t.Tuple[t.Optional[str], t.Union[exp.Table, str]]],
1456    ) -> None:
1457        table_exprs = [(gateway, exp.to_table(t)) for gateway, t in gateway_table_pairs]
1458        unique_schemas = {
1459            (gateway, t.args["db"], t.args.get("catalog"))
1460            for gateway, t in table_exprs
1461            if t and t.db
1462        }
1463
1464        def _create_schema(
1465            gateway: t.Optional[str], schema_name: str, catalog: t.Optional[str]
1466        ) -> None:
1467            schema = schema_(schema_name, catalog)
1468            logger.info("Creating schema '%s'", schema)
1469            adapter = self.get_adapter(gateway)
1470            adapter.create_schema(schema)
1471
1472        with self.concurrent_context():
1473            concurrent_apply_to_values(
1474                list(unique_schemas),
1475                lambda item: _create_schema(item[0], item[1], item[2]),
1476                self.ddl_concurrent_tasks,
1477            )
1478
1479    def get_adapter(self, gateway: t.Optional[str] = None) -> EngineAdapter:
1480        """Returns the adapter for the specified gateway or the default adapter if none is provided."""
1481        if gateway:
1482            if adapter := self.adapters.get(gateway):
1483                return adapter
1484            raise SQLMeshError(f"Gateway '{gateway}' not found in the available engine adapters.")
1485        return self.adapter
1486
1487    def _execute_create(
1488        self,
1489        snapshot: Snapshot,
1490        table_name: str,
1491        is_table_deployable: bool,
1492        deployability_index: DeployabilityIndex,
1493        create_render_kwargs: t.Dict[str, t.Any],
1494        rendered_physical_properties: t.Dict[str, exp.Expr],
1495        dry_run: bool,
1496        run_pre_post_statements: bool = True,
1497        skip_grants: bool = False,
1498    ) -> None:
1499        adapter = self.get_adapter(snapshot.model.gateway)
1500        evaluation_strategy = _evaluation_strategy(snapshot, adapter)
1501
1502        # It can still be useful for some strategies to know if the snapshot was actually deployable
1503        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
1504        is_snapshot_representative = deployability_index.is_representative(snapshot)
1505
1506        create_render_kwargs = {
1507            **create_render_kwargs,
1508            "table_mapping": {snapshot.name: table_name},
1509        }
1510        if run_pre_post_statements:
1511            evaluation_strategy.run_pre_statements(
1512                snapshot=snapshot,
1513                render_kwargs={**create_render_kwargs, "inside_transaction": True},
1514            )
1515        evaluation_strategy.create(
1516            table_name=table_name,
1517            model=snapshot.model,
1518            is_table_deployable=is_table_deployable,
1519            skip_grants=skip_grants,
1520            render_kwargs=create_render_kwargs,
1521            is_snapshot_deployable=is_snapshot_deployable,
1522            is_snapshot_representative=is_snapshot_representative,
1523            dry_run=dry_run,
1524            physical_properties=rendered_physical_properties,
1525            snapshot=snapshot,
1526            deployability_index=deployability_index,
1527        )
1528        if run_pre_post_statements:
1529            evaluation_strategy.run_post_statements(
1530                snapshot=snapshot,
1531                render_kwargs={**create_render_kwargs, "inside_transaction": True},
1532            )
1533
1534    def _can_clone(self, snapshot: Snapshot, deployability_index: DeployabilityIndex) -> bool:
1535        adapter = self.get_adapter(snapshot.model.gateway)
1536        return (
1537            snapshot.is_forward_only
1538            and snapshot.is_materialized
1539            and bool(snapshot.previous_versions)
1540            and adapter.SUPPORTS_CLONING
1541            # managed models cannot have their schema mutated because they're based on queries, so clone + alter won't work
1542            and not snapshot.is_managed
1543            and not snapshot.is_dbt_custom
1544            and not deployability_index.is_deployable(snapshot)
1545            # If the deployable table is missing we can't clone it
1546            and adapter.table_exists(snapshot.table_name())
1547        )
1548
1549    def _get_physical_data_objects(
1550        self,
1551        target_snapshots: t.Iterable[Snapshot],
1552        deployability_index: DeployabilityIndex,
1553    ) -> t.Dict[SnapshotId, DataObject]:
1554        """Returns a dictionary of snapshot IDs to existing data objects of their physical tables.
1555
1556        Args:
1557            target_snapshots: Target snapshots.
1558            deployability_index: The deployability index to determine whether to look for a deployable or
1559                a non-deployable physical table.
1560
1561        Returns:
1562            A dictionary of snapshot IDs to existing data objects of their physical tables. If the data object
1563            for a snapshot is not found, it will not be included in the dictionary.
1564        """
1565        return self._get_data_objects(
1566            target_snapshots,
1567            lambda s: exp.to_table(
1568                s.table_name(deployability_index.is_deployable(s)), dialect=s.model.dialect
1569            ),
1570        )
1571
1572    def _get_virtual_data_objects(
1573        self,
1574        target_snapshots: t.Iterable[Snapshot],
1575        environment_naming_info: EnvironmentNamingInfo,
1576    ) -> t.Dict[SnapshotId, DataObject]:
1577        """Returns a dictionary of snapshot IDs to existing data objects of their virtual views.
1578
1579        Args:
1580            target_snapshots: Target snapshots.
1581             environment_naming_info: The environment naming info of the target virtual environment.
1582
1583        Returns:
1584            A dictionary of snapshot IDs to existing data objects of their virtual views. If the data object
1585            for a snapshot is not found, it will not be included in the dictionary.
1586        """
1587
1588        def _get_view_name(s: Snapshot) -> exp.Table:
1589            adapter = (
1590                self.get_adapter(s.model_gateway)
1591                if environment_naming_info.gateway_managed
1592                else self.adapter
1593            )
1594            return exp.to_table(
1595                s.qualified_view_name.for_environment(
1596                    environment_naming_info, dialect=adapter.dialect
1597                ),
1598                dialect=adapter.dialect,
1599            )
1600
1601        return self._get_data_objects(target_snapshots, _get_view_name)
1602
1603    def _get_data_objects(
1604        self,
1605        target_snapshots: t.Iterable[Snapshot],
1606        table_name_callable: t.Callable[[Snapshot], exp.Table],
1607    ) -> t.Dict[SnapshotId, DataObject]:
1608        """Returns a dictionary of snapshot IDs to existing data objects.
1609
1610        Args:
1611            target_snapshots: Target snapshots.
1612            table_name_callable: A function that takes a snapshot and returns the table to look for.
1613
1614        Returns:
1615            A dictionary of snapshot IDs to existing data objects. If the data object for a snapshot is not found,
1616            it will not be included in the dictionary.
1617        """
1618        tables_by_gateway_and_schema: t.Dict[t.Union[str, None], t.Dict[exp.Table, set[str]]] = (
1619            defaultdict(lambda: defaultdict(set))
1620        )
1621        snapshots_by_table_name: t.Dict[exp.Table, t.Dict[str, Snapshot]] = defaultdict(dict)
1622        for snapshot in target_snapshots:
1623            if not snapshot.is_model or snapshot.is_symbolic:
1624                continue
1625            table = table_name_callable(snapshot)
1626            table_schema = d.schema_(table.db, catalog=table.catalog)
1627            tables_by_gateway_and_schema[snapshot.model_gateway][table_schema].add(table.name)
1628            snapshots_by_table_name[table_schema][table.name] = snapshot
1629
1630        def _get_data_objects_in_schema(
1631            schema: exp.Table,
1632            object_names: t.Optional[t.Set[str]] = None,
1633            gateway: t.Optional[str] = None,
1634        ) -> t.List[DataObject]:
1635            logger.info("Listing data objects in schema %s", schema.sql())
1636            return self.get_adapter(gateway).get_data_objects(
1637                schema, object_names, safe_to_cache=True
1638            )
1639
1640        with self.concurrent_context():
1641            snapshot_id_to_obj: t.Dict[SnapshotId, DataObject] = {}
1642            # A schema can be shared across multiple engines, so we need to group tables by both gateway and schema
1643            for gateway, tables_by_schema in tables_by_gateway_and_schema.items():
1644                schema_list = list(tables_by_schema.keys())
1645                results = concurrent_apply_to_values(
1646                    schema_list,
1647                    lambda s: _get_data_objects_in_schema(
1648                        schema=s, object_names=tables_by_schema.get(s), gateway=gateway
1649                    ),
1650                    self.ddl_concurrent_tasks,
1651                )
1652
1653                for schema, objs in zip(schema_list, results):
1654                    snapshots_by_name = snapshots_by_table_name.get(schema, {})
1655                    for obj in objs:
1656                        if obj.name in snapshots_by_name:
1657                            snapshot_id_to_obj[snapshots_by_name[obj.name].snapshot_id] = obj
1658
1659        return snapshot_id_to_obj
1660
1661
1662def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) -> EvaluationStrategy:
1663    klass: t.Type
1664    if snapshot.is_embedded:
1665        klass = EmbeddedStrategy
1666    elif snapshot.is_symbolic or snapshot.is_audit:
1667        klass = SymbolicStrategy
1668    elif snapshot.is_full:
1669        klass = FullRefreshStrategy
1670    elif snapshot.is_seed:
1671        klass = SeedStrategy
1672    elif snapshot.is_incremental_by_time_range:
1673        klass = IncrementalByTimeRangeStrategy
1674    elif snapshot.is_incremental_by_unique_key:
1675        klass = IncrementalByUniqueKeyStrategy
1676    elif snapshot.is_incremental_by_partition:
1677        klass = IncrementalByPartitionStrategy
1678    elif snapshot.is_incremental_unmanaged:
1679        klass = IncrementalUnmanagedStrategy
1680    elif snapshot.is_view:
1681        klass = ViewStrategy
1682    elif snapshot.is_scd_type_2:
1683        klass = SCDType2Strategy
1684    elif snapshot.is_dbt_custom:
1685        if hasattr(snapshot, "model") and isinstance(
1686            (model_kind := snapshot.model.kind), DbtCustomKind
1687        ):
1688            return DbtCustomMaterializationStrategy(
1689                adapter=adapter,
1690                materialization_name=model_kind.materialization,
1691                materialization_template=model_kind.definition,
1692            )
1693
1694        raise SQLMeshError(
1695            f"Expected DbtCustomKind for dbt custom materialization in model '{snapshot.name}'"
1696        )
1697    elif snapshot.is_custom:
1698        if snapshot.custom_materialization is None:
1699            raise SQLMeshError(
1700                f"Missing the name of a custom evaluation strategy in model '{snapshot.name}'."
1701            )
1702        _, klass = get_custom_materialization_type_or_raise(snapshot.custom_materialization)
1703        return klass(adapter)
1704    elif snapshot.is_managed:
1705        klass = EngineManagedStrategy
1706    else:
1707        raise SQLMeshError(f"Unexpected snapshot: {snapshot}")
1708
1709    return klass(adapter)
1710
1711
1712class EvaluationStrategy(abc.ABC):
1713    def __init__(self, adapter: EngineAdapter):
1714        self.adapter = adapter
1715
1716    @abc.abstractmethod
1717    def insert(
1718        self,
1719        table_name: str,
1720        query_or_df: QueryOrDF,
1721        model: Model,
1722        is_first_insert: bool,
1723        render_kwargs: t.Dict[str, t.Any],
1724        **kwargs: t.Any,
1725    ) -> None:
1726        """Inserts the given query or a DataFrame into the target table or a view.
1727
1728        Args:
1729            table_name: The name of the target table or view.
1730            query_or_df: A query or a DataFrame to insert.
1731            model: The target model.
1732            is_first_insert: Whether this is the first insert for this version of a model. This value is set to True
1733                if no data has been previously inserted into the target table, or when the entire history of the target model has
1734                been restated. Note that in the latter case, the table might contain data from previous executions, and it is the
1735                responsibility of a specific evaluation strategy to handle the truncation of the table if necessary.
1736            render_kwargs: Additional key-value arguments to pass when rendering the model's query.
1737        """
1738
1739    @abc.abstractmethod
1740    def append(
1741        self,
1742        table_name: str,
1743        query_or_df: QueryOrDF,
1744        model: Model,
1745        render_kwargs: t.Dict[str, t.Any],
1746        **kwargs: t.Any,
1747    ) -> None:
1748        """Appends the given query or a DataFrame to the existing table.
1749
1750        Args:
1751            table_name: The target table name.
1752            query_or_df: A query or a DataFrame to insert.
1753            model: The target model.
1754            render_kwargs: Additional key-value arguments to pass when rendering the model's query.
1755        """
1756
1757    @abc.abstractmethod
1758    def create(
1759        self,
1760        table_name: str,
1761        model: Model,
1762        is_table_deployable: bool,
1763        render_kwargs: t.Dict[str, t.Any],
1764        skip_grants: bool,
1765        **kwargs: t.Any,
1766    ) -> None:
1767        """Creates the target table or view.
1768
1769        Note that the intention here is to just create the table structure, data is loaded in insert() and append()
1770
1771        Args:
1772            table_name: The name of a table or a view.
1773            model: The target model.
1774            is_table_deployable: True if this creation request is for the "main" table that *might* be deployed to a production environment.
1775                False if this creation request is for the "dev preview" table. Note that this flag is not related to the DeployabilityIndex
1776                which determines if the snapshot is deployable to production or not
1777            render_kwargs: Additional key-value arguments to pass when rendering the model's query.
1778        """
1779
1780    @abc.abstractmethod
1781    def migrate(
1782        self,
1783        target_table_name: str,
1784        source_table_name: str,
1785        snapshot: Snapshot,
1786        *,
1787        ignore_destructive: bool,
1788        ignore_additive: bool,
1789        **kwargs: t.Any,
1790    ) -> None:
1791        """Migrates the target table schema so that it corresponds to the source table schema.
1792
1793        Args:
1794            target_table_name: The target table name.
1795            source_table_name: The source table name.
1796            snapshot: The target snapshot.
1797            ignore_destructive: If True, destructive changes are not created when migrating.
1798                This is used for forward-only models that are being migrated to a new version.
1799            ignore_additive: If True, additive changes are not created when migrating.
1800                This is used for forward-only models that are being migrated to a new version.
1801        """
1802
1803    @abc.abstractmethod
1804    def delete(self, name: str, **kwargs: t.Any) -> None:
1805        """Deletes a target table or a view.
1806
1807        Args:
1808            name: The name of a table or a view.
1809        """
1810
1811    @abc.abstractmethod
1812    def promote(
1813        self,
1814        table_name: str,
1815        view_name: str,
1816        model: Model,
1817        environment: str,
1818        **kwargs: t.Any,
1819    ) -> None:
1820        """Updates the target view to point to the target table.
1821
1822        Args:
1823            table_name: The name of a table in the physical layer that is being promoted.
1824            view_name: The name of the target view in the virtual layer.
1825            model: The model that is being promoted.
1826            environment: The name of the target environment.
1827        """
1828
1829    @abc.abstractmethod
1830    def demote(self, view_name: str, **kwargs: t.Any) -> None:
1831        """Deletes the target view in the virtual layer.
1832
1833        Args:
1834            view_name: The name of the target view in the virtual layer.
1835        """
1836
1837    @abc.abstractmethod
1838    def run_pre_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
1839        """Executes the snapshot's pre statements.
1840
1841        Args:
1842            snapshot: The target snapshot.
1843            render_kwargs: Additional key-value arguments to pass when rendering the statements.
1844        """
1845
1846    @abc.abstractmethod
1847    def run_post_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
1848        """Executes the snapshot's post statements.
1849
1850        Args:
1851            snapshot: The target snapshot.
1852            render_kwargs: Additional key-value arguments to pass when rendering the statements.
1853        """
1854
1855    def _apply_grants(
1856        self,
1857        model: Model,
1858        table_name: str,
1859        target_layer: GrantsTargetLayer,
1860        is_snapshot_deployable: bool = False,
1861    ) -> None:
1862        """Apply grants for a model if grants are configured.
1863
1864        This method provides consistent grants application across all evaluation strategies.
1865        It ensures that whenever a physical database object (table, view, materialized view)
1866        is created or modified, the appropriate grants are applied.
1867
1868        Args:
1869            model: The SQLMesh model containing grants configuration
1870            table_name: The target table/view name to apply grants to
1871            target_layer: The grants application layer (physical or virtual)
1872            is_snapshot_deployable: Whether the snapshot is deployable (targeting production)
1873        """
1874        grants_config = model.grants
1875        if grants_config is None:
1876            return
1877
1878        if not self.adapter.SUPPORTS_GRANTS:
1879            logger.warning(
1880                f"Engine {self.adapter.__class__.__name__} does not support grants. "
1881                f"Skipping grants application for model {model.name}"
1882            )
1883            return
1884
1885        model_grants_target_layer = model.grants_target_layer
1886        deployable_vde_dev_only = (
1887            is_snapshot_deployable and model.virtual_environment_mode.is_dev_only
1888        )
1889
1890        # table_type is always a VIEW in the virtual layer unless model is deployable and VDE is dev_only
1891        # in which case we fall back to the model's model_grants_table_type
1892        if target_layer == GrantsTargetLayer.VIRTUAL and not deployable_vde_dev_only:
1893            model_grants_table_type = DataObjectType.VIEW
1894        else:
1895            model_grants_table_type = model.grants_table_type
1896
1897        if (
1898            model_grants_target_layer.is_all
1899            or model_grants_target_layer == target_layer
1900            # Always apply grants in production when VDE is dev_only regardless of target_layer
1901            # since only physical tables are created in production
1902            or deployable_vde_dev_only
1903        ):
1904            logger.info(f"Applying grants for model {model.name} to table {table_name}")
1905            self.adapter.sync_grants_config(
1906                exp.to_table(table_name, dialect=self.adapter.dialect),
1907                grants_config,
1908                model_grants_table_type,
1909            )
1910        else:
1911            logger.debug(
1912                f"Skipping grants application for model {model.name} in {target_layer} layer"
1913            )
1914
1915
1916class SymbolicStrategy(EvaluationStrategy):
1917    def insert(
1918        self,
1919        table_name: str,
1920        query_or_df: QueryOrDF,
1921        model: Model,
1922        is_first_insert: bool,
1923        render_kwargs: t.Dict[str, t.Any],
1924        **kwargs: t.Any,
1925    ) -> None:
1926        pass
1927
1928    def append(
1929        self,
1930        table_name: str,
1931        query_or_df: QueryOrDF,
1932        model: Model,
1933        render_kwargs: t.Dict[str, t.Any],
1934        **kwargs: t.Any,
1935    ) -> None:
1936        pass
1937
1938    def create(
1939        self,
1940        table_name: str,
1941        model: Model,
1942        is_table_deployable: bool,
1943        render_kwargs: t.Dict[str, t.Any],
1944        skip_grants: bool,
1945        **kwargs: t.Any,
1946    ) -> None:
1947        pass
1948
1949    def migrate(
1950        self,
1951        target_table_name: str,
1952        source_table_name: str,
1953        snapshot: Snapshot,
1954        *,
1955        ignore_destructive: bool,
1956        ignore_additive: bool,
1957        **kwarg: t.Any,
1958    ) -> None:
1959        pass
1960
1961    def delete(self, name: str, **kwargs: t.Any) -> None:
1962        pass
1963
1964    def promote(
1965        self,
1966        table_name: str,
1967        view_name: str,
1968        model: Model,
1969        environment: str,
1970        **kwargs: t.Any,
1971    ) -> None:
1972        pass
1973
1974    def demote(self, view_name: str, **kwargs: t.Any) -> None:
1975        pass
1976
1977    def run_pre_statements(self, snapshot: Snapshot, render_kwargs: t.Dict[str, t.Any]) -> None:
1978        pass
1979
1980    def run_post_statements(self, snapshot: Snapshot, render_kwargs: t.Dict[str, t.Any]) -> None:
1981        pass
1982
1983
1984class EmbeddedStrategy(SymbolicStrategy):
1985    def promote(
1986        self,
1987        table_name: str,
1988        view_name: str,
1989        model: Model,
1990        environment: str,
1991        **kwargs: t.Any,
1992    ) -> None:
1993        logger.info("Dropping view '%s' for non-materialized table", view_name)
1994        self.adapter.drop_view(view_name, cascade=False)
1995
1996
1997class PromotableStrategy(EvaluationStrategy, abc.ABC):
1998    def promote(
1999        self,
2000        table_name: str,
2001        view_name: str,
2002        model: Model,
2003        environment: str,
2004        **kwargs: t.Any,
2005    ) -> None:
2006        is_prod = environment == c.PROD
2007        logger.info("Updating view '%s' to point at table '%s'", view_name, table_name)
2008        render_kwargs: t.Dict[str, t.Any] = dict(
2009            start=kwargs.get("start"),
2010            end=kwargs.get("end"),
2011            execution_time=kwargs.get("execution_time"),
2012            engine_adapter=kwargs.get("engine_adapter"),
2013            snapshots=kwargs.get("snapshots"),
2014            deployability_index=kwargs.get("deployability_index"),
2015            table_mapping=kwargs.get("table_mapping"),
2016            runtime_stage=kwargs.get("runtime_stage"),
2017        )
2018        self.adapter.create_view(
2019            view_name,
2020            exp.select("*").from_(table_name, dialect=self.adapter.dialect),
2021            table_description=model.description if is_prod else None,
2022            column_descriptions=model.column_descriptions if is_prod else None,
2023            view_properties=model.render_virtual_properties(**render_kwargs),
2024        )
2025
2026        snapshot = kwargs.get("snapshot")
2027        deployability_index = kwargs.get("deployability_index")
2028        is_snapshot_deployable = (
2029            deployability_index.is_deployable(snapshot)
2030            if snapshot and deployability_index
2031            else False
2032        )
2033
2034        # Apply grants to the virtual layer (view) after promotion
2035        self._apply_grants(model, view_name, GrantsTargetLayer.VIRTUAL, is_snapshot_deployable)
2036
2037    def demote(self, view_name: str, **kwargs: t.Any) -> None:
2038        logger.info("Dropping view '%s'", view_name)
2039        self.adapter.drop_view(view_name, cascade=False)
2040
2041    def run_pre_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
2042        self.adapter.execute(snapshot.model.render_pre_statements(**render_kwargs))
2043
2044    def run_post_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
2045        self.adapter.execute(snapshot.model.render_post_statements(**render_kwargs))
2046
2047
2048def _adjust_physical_properties_for_engine(
2049    adapter: EngineAdapter,
2050    model: Model,
2051    physical_properties: t.Optional[t.Dict[str, t.Any]],
2052) -> t.Dict[str, t.Any]:
2053    """Let the target engine adjust/validate physical properties for an incremental model.
2054
2055    The generic responsibility here is to determine, from the model kind, whether the table will
2056    be the target of DELETE/MERGE statements (vs. append-only INSERTs) and whether its unique_key
2057    may be promoted to an engine-specific key. The engine adapter decides what, if anything, to do
2058    with that information (see ``EngineAdapter.adjust_physical_properties_for_incremental``).
2059    """
2060    kind = model.kind
2061
2062    # Only incremental kinds that issue DELETE/MERGE need a delete-capable table. Append-only
2063    # INCREMENTAL_UNMANAGED (insert_overwrite=False) only does INSERT, so it does not.
2064    requires_delete_capable_table = (
2065        kind.is_incremental_by_time_range
2066        or kind.is_incremental_by_unique_key
2067        or kind.is_incremental_by_partition
2068        or kind.is_scd_type_2
2069        or (isinstance(kind, IncrementalUnmanagedKind) and kind.insert_overwrite)
2070    )
2071
2072    return adapter.adjust_physical_properties_for_incremental(
2073        dict(physical_properties or {}),
2074        requires_delete_capable_table=requires_delete_capable_table,
2075        unique_key=model.unique_key if kind.is_incremental_by_unique_key else None,
2076        model_name=model.name,
2077    )
2078
2079
2080class MaterializableStrategy(PromotableStrategy, abc.ABC):
2081    def create(
2082        self,
2083        table_name: str,
2084        model: Model,
2085        is_table_deployable: bool,
2086        render_kwargs: t.Dict[str, t.Any],
2087        skip_grants: bool,
2088        **kwargs: t.Any,
2089    ) -> None:
2090        ctas_query = model.ctas_query(**render_kwargs)
2091        physical_properties = _adjust_physical_properties_for_engine(
2092            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
2093        )
2094
2095        logger.info("Creating table '%s'", table_name)
2096        if model.annotated:
2097            self.adapter.create_table(
2098                table_name,
2099                target_columns_to_types=model.columns_to_types_or_raise,
2100                table_format=model.table_format,
2101                storage_format=model.storage_format,
2102                partitioned_by=model.partitioned_by,
2103                partition_interval_unit=model.partition_interval_unit,
2104                clustered_by=model.clustered_by,
2105                table_properties=physical_properties,
2106                table_description=model.description if is_table_deployable else None,
2107                column_descriptions=model.column_descriptions if is_table_deployable else None,
2108            )
2109
2110            # If we create both temp and prod tables, we need to make sure that we dry run once.
2111            dry_run = kwargs.get("dry_run", True) or not is_table_deployable
2112
2113            # Only sql models have queries that can be tested.
2114            # We also need to make sure that we don't dry run on Redshift because its planner / optimizer sometimes
2115            # breaks on our CTAS queries due to us relying on the WHERE FALSE LIMIT 0 combo.
2116            if model.is_sql and dry_run and self.adapter.dialect != "redshift":
2117                logger.info("Dry running model '%s'", model.name)
2118                self.adapter.fetchall(ctas_query)
2119        else:
2120            self.adapter.ctas(
2121                table_name,
2122                ctas_query,
2123                model.columns_to_types,
2124                table_format=model.table_format,
2125                storage_format=model.storage_format,
2126                partitioned_by=model.partitioned_by,
2127                partition_interval_unit=model.partition_interval_unit,
2128                clustered_by=model.clustered_by,
2129                table_properties=physical_properties,
2130                table_description=model.description if is_table_deployable else None,
2131                column_descriptions=model.column_descriptions if is_table_deployable else None,
2132            )
2133
2134        # Apply grants after table creation (unless explicitly skipped by caller)
2135        if not skip_grants:
2136            is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2137            self._apply_grants(
2138                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2139            )
2140
2141    def migrate(
2142        self,
2143        target_table_name: str,
2144        source_table_name: str,
2145        snapshot: Snapshot,
2146        *,
2147        ignore_destructive: bool,
2148        ignore_additive: bool,
2149        **kwargs: t.Any,
2150    ) -> None:
2151        logger.info(f"Altering table '{target_table_name}'")
2152        alter_operations = self.adapter.get_alter_operations(
2153            target_table_name,
2154            source_table_name,
2155            ignore_destructive=ignore_destructive,
2156            ignore_additive=ignore_additive,
2157        )
2158        _check_destructive_schema_change(
2159            snapshot, alter_operations, kwargs["allow_destructive_snapshots"]
2160        )
2161        _check_additive_schema_change(
2162            snapshot, alter_operations, kwargs["allow_additive_snapshots"]
2163        )
2164        self.adapter.alter_table(alter_operations)
2165
2166        # Apply grants after schema migration
2167        deployability_index = kwargs.get("deployability_index")
2168        is_snapshot_deployable = (
2169            deployability_index.is_deployable(snapshot) if deployability_index else False
2170        )
2171        self._apply_grants(
2172            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2173        )
2174
2175    def delete(self, name: str, **kwargs: t.Any) -> None:
2176        _check_table_db_is_physical_schema(name, kwargs["physical_schema"])
2177        self.adapter.drop_table(name, cascade=kwargs.pop("cascade", False))
2178        logger.info("Dropped table '%s'", name)
2179
2180    def _replace_query_for_model(
2181        self,
2182        model: Model,
2183        name: str,
2184        query_or_df: QueryOrDF,
2185        render_kwargs: t.Dict[str, t.Any],
2186        skip_grants: bool = False,
2187        **kwargs: t.Any,
2188    ) -> None:
2189        """Replaces the table for the given model.
2190
2191        Args:
2192            model: The target model.
2193            name: The name of the target table.
2194            query_or_df: The query or DataFrame to replace the target table with.
2195        """
2196        if (model.is_seed or model.kind.is_full) and model.annotated:
2197            columns_to_types = model.columns_to_types_or_raise
2198            source_columns: t.Optional[t.List[str]] = list(columns_to_types)
2199        else:
2200            try:
2201                # Source columns from the underlying table to prevent unintentional table schema changes during restatement of incremental models.
2202                columns_to_types, source_columns = self._get_target_and_source_columns(
2203                    model, name, render_kwargs, force_get_columns_from_target=True
2204                )
2205            except Exception:
2206                columns_to_types, source_columns = None, None
2207
2208        physical_properties = _adjust_physical_properties_for_engine(
2209            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
2210        )
2211        self.adapter.replace_query(
2212            name,
2213            query_or_df,
2214            table_format=model.table_format,
2215            storage_format=model.storage_format,
2216            partitioned_by=model.partitioned_by,
2217            partition_interval_unit=model.partition_interval_unit,
2218            clustered_by=model.clustered_by,
2219            table_properties=physical_properties,
2220            table_description=model.description,
2221            column_descriptions=model.column_descriptions,
2222            target_columns_to_types=columns_to_types,
2223            source_columns=source_columns,
2224        )
2225
2226        # Apply grants after table replacement (unless explicitly skipped by caller)
2227        if not skip_grants:
2228            is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2229            self._apply_grants(model, name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
2230
2231    def _get_target_and_source_columns(
2232        self,
2233        model: Model,
2234        table_name: str,
2235        render_kwargs: t.Dict[str, t.Any],
2236        force_get_columns_from_target: bool = False,
2237    ) -> t.Tuple[t.Dict[str, exp.DataType], t.Optional[t.List[str]]]:
2238        if force_get_columns_from_target:
2239            target_column_to_types = self.adapter.columns(table_name)
2240        else:
2241            target_column_to_types = (
2242                model.columns_to_types  # type: ignore
2243                if model.annotated
2244                and not model.on_destructive_change.is_ignore
2245                and not model.on_additive_change.is_ignore
2246                else self.adapter.columns(table_name)
2247            )
2248        assert target_column_to_types is not None
2249        if model.on_destructive_change.is_ignore or model.on_additive_change.is_ignore:
2250            # We need to identify the columns that are only in the source so we create an empty table with
2251            # the user query to determine that
2252            temp_table_name = exp.table_(
2253                "diff",
2254                db=model.physical_schema,
2255            )
2256            with self.adapter.temp_table(
2257                model.ctas_query(**render_kwargs), name=temp_table_name
2258            ) as temp_table:
2259                source_columns = list(self.adapter.columns(temp_table))
2260        else:
2261            source_columns = None
2262        return target_column_to_types, source_columns
2263
2264
2265class IncrementalStrategy(MaterializableStrategy, abc.ABC):
2266    def append(
2267        self,
2268        table_name: str,
2269        query_or_df: QueryOrDF,
2270        model: Model,
2271        render_kwargs: t.Dict[str, t.Any],
2272        **kwargs: t.Any,
2273    ) -> None:
2274        columns_to_types, source_columns = self._get_target_and_source_columns(
2275            model, table_name, render_kwargs=render_kwargs
2276        )
2277        self.adapter.insert_append(
2278            table_name,
2279            query_or_df,
2280            target_columns_to_types=columns_to_types,
2281            source_columns=source_columns,
2282        )
2283
2284
2285class IncrementalByPartitionStrategy(IncrementalStrategy):
2286    def insert(
2287        self,
2288        table_name: str,
2289        query_or_df: QueryOrDF,
2290        model: Model,
2291        is_first_insert: bool,
2292        render_kwargs: t.Dict[str, t.Any],
2293        **kwargs: t.Any,
2294    ) -> None:
2295        if is_first_insert:
2296            self._replace_query_for_model(model, table_name, query_or_df, render_kwargs, **kwargs)
2297        else:
2298            columns_to_types, source_columns = self._get_target_and_source_columns(
2299                model, table_name, render_kwargs=render_kwargs
2300            )
2301            self.adapter.insert_overwrite_by_partition(
2302                table_name,
2303                query_or_df,
2304                partitioned_by=model.partitioned_by,
2305                target_columns_to_types=columns_to_types,
2306                source_columns=source_columns,
2307            )
2308
2309
2310class IncrementalByTimeRangeStrategy(IncrementalStrategy):
2311    def insert(
2312        self,
2313        table_name: str,
2314        query_or_df: QueryOrDF,
2315        model: Model,
2316        is_first_insert: bool,
2317        render_kwargs: t.Dict[str, t.Any],
2318        **kwargs: t.Any,
2319    ) -> None:
2320        assert model.time_column
2321        columns_to_types, source_columns = self._get_target_and_source_columns(
2322            model, table_name, render_kwargs=render_kwargs
2323        )
2324        self.adapter.insert_overwrite_by_time_partition(
2325            table_name,
2326            query_or_df,
2327            time_formatter=model.convert_to_time_column,
2328            time_column=model.time_column,
2329            target_columns_to_types=columns_to_types,
2330            source_columns=source_columns,
2331            **kwargs,
2332        )
2333
2334
2335class IncrementalByUniqueKeyStrategy(IncrementalStrategy):
2336    def insert(
2337        self,
2338        table_name: str,
2339        query_or_df: QueryOrDF,
2340        model: Model,
2341        is_first_insert: bool,
2342        render_kwargs: t.Dict[str, t.Any],
2343        **kwargs: t.Any,
2344    ) -> None:
2345        if is_first_insert:
2346            self._replace_query_for_model(model, table_name, query_or_df, render_kwargs, **kwargs)
2347        else:
2348            columns_to_types, source_columns = self._get_target_and_source_columns(
2349                model,
2350                table_name,
2351                render_kwargs=render_kwargs,
2352            )
2353            physical_properties = _adjust_physical_properties_for_engine(
2354                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
2355            )
2356            self.adapter.merge(
2357                table_name,
2358                query_or_df,
2359                target_columns_to_types=columns_to_types,
2360                unique_key=model.unique_key,
2361                when_matched=model.when_matched,
2362                merge_filter=model.render_merge_filter(
2363                    start=kwargs.get("start"),
2364                    end=kwargs.get("end"),
2365                    execution_time=kwargs.get("execution_time"),
2366                ),
2367                physical_properties=physical_properties,
2368                source_columns=source_columns,
2369            )
2370
2371    def append(
2372        self,
2373        table_name: str,
2374        query_or_df: QueryOrDF,
2375        model: Model,
2376        render_kwargs: t.Dict[str, t.Any],
2377        **kwargs: t.Any,
2378    ) -> None:
2379        columns_to_types, source_columns = self._get_target_and_source_columns(
2380            model, table_name, render_kwargs=render_kwargs
2381        )
2382        physical_properties = _adjust_physical_properties_for_engine(
2383            self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
2384        )
2385        self.adapter.merge(
2386            table_name,
2387            query_or_df,
2388            target_columns_to_types=columns_to_types,
2389            unique_key=model.unique_key,
2390            when_matched=model.when_matched,
2391            merge_filter=model.render_merge_filter(
2392                start=kwargs.get("start"),
2393                end=kwargs.get("end"),
2394                execution_time=kwargs.get("execution_time"),
2395            ),
2396            physical_properties=physical_properties,
2397            source_columns=source_columns,
2398        )
2399
2400
2401class IncrementalUnmanagedStrategy(IncrementalStrategy):
2402    def append(
2403        self,
2404        table_name: str,
2405        query_or_df: QueryOrDF,
2406        model: Model,
2407        render_kwargs: t.Dict[str, t.Any],
2408        **kwargs: t.Any,
2409    ) -> None:
2410        columns_to_types, source_columns = self._get_target_and_source_columns(
2411            model, table_name, render_kwargs=render_kwargs
2412        )
2413        self.adapter.insert_append(
2414            table_name,
2415            query_or_df,
2416            target_columns_to_types=columns_to_types,
2417            source_columns=source_columns,
2418        )
2419
2420    def insert(
2421        self,
2422        table_name: str,
2423        query_or_df: QueryOrDF,
2424        model: Model,
2425        is_first_insert: bool,
2426        render_kwargs: t.Dict[str, t.Any],
2427        **kwargs: t.Any,
2428    ) -> None:
2429        if is_first_insert:
2430            return self._replace_query_for_model(
2431                model, table_name, query_or_df, render_kwargs, **kwargs
2432            )
2433        if isinstance(model.kind, IncrementalUnmanagedKind) and model.kind.insert_overwrite:
2434            columns_to_types, source_columns = self._get_target_and_source_columns(
2435                model,
2436                table_name,
2437                render_kwargs=render_kwargs,
2438            )
2439
2440            return self.adapter.insert_overwrite_by_partition(
2441                table_name,
2442                query_or_df,
2443                model.partitioned_by,
2444                target_columns_to_types=columns_to_types,
2445                source_columns=source_columns,
2446            )
2447        return self.append(
2448            table_name,
2449            query_or_df,
2450            model,
2451            render_kwargs=render_kwargs,
2452            **kwargs,
2453        )
2454
2455
2456class FullRefreshStrategy(MaterializableStrategy):
2457    def append(
2458        self,
2459        table_name: str,
2460        query_or_df: QueryOrDF,
2461        model: Model,
2462        render_kwargs: t.Dict[str, t.Any],
2463        **kwargs: t.Any,
2464    ) -> None:
2465        self.adapter.insert_append(
2466            table_name,
2467            query_or_df,
2468            target_columns_to_types=model.columns_to_types,
2469        )
2470
2471    def insert(
2472        self,
2473        table_name: str,
2474        query_or_df: QueryOrDF,
2475        model: Model,
2476        is_first_insert: bool,
2477        render_kwargs: t.Dict[str, t.Any],
2478        **kwargs: t.Any,
2479    ) -> None:
2480        self._replace_query_for_model(model, table_name, query_or_df, render_kwargs, **kwargs)
2481
2482
2483class SeedStrategy(MaterializableStrategy):
2484    def create(
2485        self,
2486        table_name: str,
2487        model: Model,
2488        is_table_deployable: bool,
2489        render_kwargs: t.Dict[str, t.Any],
2490        skip_grants: bool,
2491        **kwargs: t.Any,
2492    ) -> None:
2493        model = t.cast(SeedModel, model)
2494        if not model.is_hydrated and self.adapter.table_exists(table_name):
2495            # This likely means that the table was created and populated previously, but the evaluation stage
2496            # failed before the interval could be added for this model.
2497            logger.warning(
2498                "Seed model '%s' is not hydrated, but the table '%s' exists. Skipping creation",
2499                model.name,
2500                table_name,
2501            )
2502            return
2503
2504        super().create(
2505            table_name,
2506            model,
2507            is_table_deployable,
2508            render_kwargs,
2509            skip_grants=True,  # Skip grants; they're applied after data insertion
2510            **kwargs,
2511        )
2512        # For seeds we insert data at the time of table creation.
2513        try:
2514            for index, df in enumerate(model.render_seed()):
2515                if index == 0:
2516                    self._replace_query_for_model(
2517                        model,
2518                        table_name,
2519                        df,
2520                        render_kwargs,
2521                        skip_grants=True,  # Skip grants; they're applied after data insertion
2522                        **kwargs,
2523                    )
2524                else:
2525                    self.adapter.insert_append(
2526                        table_name, df, target_columns_to_types=model.columns_to_types
2527                    )
2528
2529            if not skip_grants:
2530                # Apply grants after seed table creation and data insertion
2531                is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2532                self._apply_grants(
2533                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2534                )
2535        except Exception:
2536            self.adapter.drop_table(table_name)
2537            raise
2538
2539    def migrate(
2540        self,
2541        target_table_name: str,
2542        source_table_name: str,
2543        snapshot: Snapshot,
2544        *,
2545        ignore_destructive: bool,
2546        ignore_additive: bool,
2547        **kwargs: t.Any,
2548    ) -> None:
2549        raise NotImplementedError("Seeds do not support migrations.")
2550
2551    def insert(
2552        self,
2553        table_name: str,
2554        query_or_df: QueryOrDF,
2555        model: Model,
2556        is_first_insert: bool,
2557        render_kwargs: t.Dict[str, t.Any],
2558        **kwargs: t.Any,
2559    ) -> None:
2560        # Data has already been inserted at the time of table creation.
2561        pass
2562
2563    def append(
2564        self,
2565        table_name: str,
2566        query_or_df: QueryOrDF,
2567        model: Model,
2568        render_kwargs: t.Dict[str, t.Any],
2569        **kwargs: t.Any,
2570    ) -> None:
2571        # Data has already been inserted at the time of table creation.
2572        pass
2573
2574
2575class SCDType2Strategy(IncrementalStrategy):
2576    def create(
2577        self,
2578        table_name: str,
2579        model: Model,
2580        is_table_deployable: bool,
2581        render_kwargs: t.Dict[str, t.Any],
2582        skip_grants: bool,
2583        **kwargs: t.Any,
2584    ) -> None:
2585        assert isinstance(model.kind, (SCDType2ByTimeKind, SCDType2ByColumnKind))
2586        if model.annotated:
2587            logger.info("Creating table '%s'", table_name)
2588            columns_to_types = model.columns_to_types_or_raise
2589            if isinstance(model.kind, SCDType2ByTimeKind):
2590                columns_to_types[model.kind.updated_at_name.name] = model.kind.time_data_type
2591            physical_properties = _adjust_physical_properties_for_engine(
2592                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
2593            )
2594            self.adapter.create_table(
2595                table_name,
2596                target_columns_to_types=columns_to_types,
2597                table_format=model.table_format,
2598                storage_format=model.storage_format,
2599                partitioned_by=model.partitioned_by,
2600                partition_interval_unit=model.partition_interval_unit,
2601                clustered_by=model.clustered_by,
2602                table_properties=physical_properties,
2603                table_description=model.description if is_table_deployable else None,
2604                column_descriptions=model.column_descriptions if is_table_deployable else None,
2605            )
2606        else:
2607            # We assume that the data type for `updated_at_name` matches the data type that is defined for
2608            # `time_data_type`. If that isn't the case, then the user might get an error about not being able
2609            # to do comparisons across different data types
2610            super().create(
2611                table_name,
2612                model,
2613                is_table_deployable,
2614                render_kwargs,
2615                skip_grants,
2616                **kwargs,
2617            )
2618
2619        if not skip_grants:
2620            # Apply grants after SCD Type 2 table creation
2621            is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2622            self._apply_grants(
2623                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2624            )
2625
2626    def insert(
2627        self,
2628        table_name: str,
2629        query_or_df: QueryOrDF,
2630        model: Model,
2631        is_first_insert: bool,
2632        render_kwargs: t.Dict[str, t.Any],
2633        **kwargs: t.Any,
2634    ) -> None:
2635        # Source columns from the underlying table to prevent unintentional table schema changes during restatement of incremental models.
2636        columns_to_types, source_columns = self._get_target_and_source_columns(
2637            model,
2638            table_name,
2639            render_kwargs=render_kwargs,
2640            force_get_columns_from_target=True,
2641        )
2642        if isinstance(model.kind, SCDType2ByTimeKind):
2643            self.adapter.scd_type_2_by_time(
2644                target_table=table_name,
2645                source_table=query_or_df,
2646                unique_key=model.unique_key,
2647                valid_from_col=model.kind.valid_from_name,
2648                valid_to_col=model.kind.valid_to_name,
2649                execution_time=kwargs["execution_time"],
2650                updated_at_col=model.kind.updated_at_name,
2651                invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
2652                updated_at_as_valid_from=model.kind.updated_at_as_valid_from,
2653                target_columns_to_types=columns_to_types,
2654                table_format=model.table_format,
2655                table_description=model.description,
2656                column_descriptions=model.column_descriptions,
2657                truncate=is_first_insert,
2658                source_columns=source_columns,
2659                storage_format=model.storage_format,
2660                partitioned_by=model.partitioned_by,
2661                partition_interval_unit=model.partition_interval_unit,
2662                clustered_by=model.clustered_by,
2663                table_properties=kwargs.get("physical_properties", model.physical_properties),
2664            )
2665        elif isinstance(model.kind, SCDType2ByColumnKind):
2666            self.adapter.scd_type_2_by_column(
2667                target_table=table_name,
2668                source_table=query_or_df,
2669                unique_key=model.unique_key,
2670                valid_from_col=model.kind.valid_from_name,
2671                valid_to_col=model.kind.valid_to_name,
2672                execution_time=model.kind.updated_at_name or kwargs["execution_time"],
2673                check_columns=model.kind.columns,
2674                invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
2675                execution_time_as_valid_from=model.kind.execution_time_as_valid_from,
2676                target_columns_to_types=columns_to_types,
2677                table_format=model.table_format,
2678                table_description=model.description,
2679                column_descriptions=model.column_descriptions,
2680                truncate=is_first_insert,
2681                source_columns=source_columns,
2682                storage_format=model.storage_format,
2683                partitioned_by=model.partitioned_by,
2684                partition_interval_unit=model.partition_interval_unit,
2685                clustered_by=model.clustered_by,
2686                table_properties=kwargs.get("physical_properties", model.physical_properties),
2687            )
2688        else:
2689            raise SQLMeshError(
2690                f"Unexpected SCD Type 2 kind: {model.kind}. This is not expected and please report this as a bug."
2691            )
2692
2693        # Apply grants after SCD Type 2 table recreation
2694        is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2695        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
2696
2697    def append(
2698        self,
2699        table_name: str,
2700        query_or_df: QueryOrDF,
2701        model: Model,
2702        render_kwargs: t.Dict[str, t.Any],
2703        **kwargs: t.Any,
2704    ) -> None:
2705        return self.insert(
2706            table_name,
2707            query_or_df,
2708            model,
2709            is_first_insert=False,
2710            render_kwargs=render_kwargs,
2711            **kwargs,
2712        )
2713
2714
2715class ViewStrategy(PromotableStrategy):
2716    def insert(
2717        self,
2718        table_name: str,
2719        query_or_df: QueryOrDF,
2720        model: Model,
2721        is_first_insert: bool,
2722        render_kwargs: t.Dict[str, t.Any],
2723        **kwargs: t.Any,
2724    ) -> None:
2725        # We should recreate MVs across supported engines (Snowflake, BigQuery etc) because
2726        # if upstream tables were recreated (e.g FULL models), the MVs would be silently invalidated.
2727        # The only exception to that rule is RisingWave which doesn't support CREATE OR REPLACE, so upstream
2728        # models don't recreate their physical tables for the MVs to be invalidated.
2729        # However, even for RW we still want to recreate MVs to avoid stale references, as is the case with normal views.
2730        # The flag is_first_insert is used for that matter as a signal to recreate the MV if the snapshot's intervals
2731        # have been cleared by `should_force_rebuild`
2732        is_materialized_view = self._is_materialized_view(model)
2733        must_recreate_view = not self.adapter.HAS_VIEW_BINDING or (
2734            is_materialized_view and is_first_insert
2735        )
2736
2737        # Some engines (e.g. StarRocks) maintain materialized views automatically (via auto/scheduled
2738        # REFRESH) and can only recreate them via a destructive DROP + CREATE, which deletes the
2739        # materialized data and forces a full rebuild. For those, an existing MV must not be recreated
2740        # on routine evaluation (e.g. every `sqlmesh run`); only build it on the first insert (a new
2741        # version) or when a rebuild is explicitly forced (intervals cleared by `should_force_rebuild`,
2742        # which sets `is_first_insert`).
2743        if (
2744            is_materialized_view
2745            and not is_first_insert
2746            and not self.adapter.RECREATE_MATERIALIZED_VIEW_ON_EVALUATION
2747        ):
2748            must_recreate_view = False
2749
2750        if self.adapter.table_exists(table_name) and not must_recreate_view:
2751            logger.info("Skipping creation of the view '%s'", table_name)
2752            return
2753
2754        logger.info("Replacing view '%s'", table_name)
2755        materialized_properties = None
2756        if is_materialized_view:
2757            materialized_properties = {
2758                "partitioned_by": model.partitioned_by,
2759                "partition_interval_unit": model.partition_interval_unit,
2760                "clustered_by": model.clustered_by,
2761                "has_audits": bool(model.audits_with_args),
2762            }
2763        self.adapter.create_view(
2764            table_name,
2765            query_or_df,
2766            model.columns_to_types,
2767            replace=must_recreate_view,
2768            materialized=is_materialized_view,
2769            materialized_properties=materialized_properties,
2770            view_properties=kwargs.get("physical_properties", model.physical_properties),
2771            table_description=model.description,
2772            column_descriptions=model.column_descriptions,
2773        )
2774
2775        # Apply grants after view creation / replacement
2776        is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2777        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
2778
2779    def append(
2780        self,
2781        table_name: str,
2782        query_or_df: QueryOrDF,
2783        model: Model,
2784        render_kwargs: t.Dict[str, t.Any],
2785        **kwargs: t.Any,
2786    ) -> None:
2787        raise ConfigError(f"Cannot append to a view '{table_name}'.")
2788
2789    def create(
2790        self,
2791        table_name: str,
2792        model: Model,
2793        is_table_deployable: bool,
2794        render_kwargs: t.Dict[str, t.Any],
2795        skip_grants: bool,
2796        **kwargs: t.Any,
2797    ) -> None:
2798        is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
2799
2800        if self.adapter.table_exists(table_name):
2801            # Make sure we don't recreate the view to prevent deletion of downstream views in engines with no late
2802            # binding support (because of DROP CASCADE).
2803            logger.info("View '%s' already exists", table_name)
2804
2805            if not skip_grants:
2806                # Always apply grants when present, even if view exists, to handle grants updates
2807                self._apply_grants(
2808                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2809                )
2810            return
2811
2812        logger.info("Creating view '%s'", table_name)
2813        materialized = self._is_materialized_view(model)
2814        materialized_properties = None
2815        if materialized:
2816            materialized_properties = {
2817                "partitioned_by": model.partitioned_by,
2818                "clustered_by": model.clustered_by,
2819                "partition_interval_unit": model.partition_interval_unit,
2820                "has_audits": bool(model.audits_with_args),
2821            }
2822        self.adapter.create_view(
2823            table_name,
2824            model.render_query_or_raise(**render_kwargs),
2825            # Make sure we never replace the view during creation to avoid race conditions in engines with no late binding support.
2826            replace=False,
2827            materialized=self._is_materialized_view(model),
2828            materialized_properties=materialized_properties,
2829            view_properties=kwargs.get("physical_properties", model.physical_properties),
2830            table_description=model.description if is_table_deployable else None,
2831            column_descriptions=model.column_descriptions if is_table_deployable else None,
2832        )
2833
2834        if not skip_grants:
2835            # Apply grants after view creation
2836            self._apply_grants(
2837                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2838            )
2839
2840    def migrate(
2841        self,
2842        target_table_name: str,
2843        source_table_name: str,
2844        snapshot: Snapshot,
2845        *,
2846        ignore_destructive: bool,
2847        ignore_additive: bool,
2848        **kwargs: t.Any,
2849    ) -> None:
2850        logger.info("Migrating view '%s'", target_table_name)
2851        model = snapshot.model
2852        render_kwargs = dict(
2853            execution_time=now(), snapshots=kwargs["snapshots"], engine_adapter=self.adapter
2854        )
2855
2856        is_materialized_view = self._is_materialized_view(model)
2857        materialized_properties = None
2858        if is_materialized_view:
2859            materialized_properties = {
2860                "partitioned_by": model.partitioned_by,
2861                "clustered_by": model.clustered_by,
2862                "partition_interval_unit": model.partition_interval_unit,
2863                "has_audits": bool(model.audits_with_args),
2864            }
2865
2866        self.adapter.create_view(
2867            target_table_name,
2868            model.render_query_or_raise(**render_kwargs),
2869            model.columns_to_types,
2870            materialized=is_materialized_view,
2871            materialized_properties=materialized_properties,
2872            view_properties=model.render_physical_properties(**render_kwargs),
2873            table_description=model.description,
2874            column_descriptions=model.column_descriptions,
2875        )
2876
2877        # Apply grants after view migration
2878        deployability_index = kwargs.get("deployability_index")
2879        is_snapshot_deployable = (
2880            deployability_index.is_deployable(snapshot) if deployability_index else False
2881        )
2882        self._apply_grants(
2883            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
2884        )
2885
2886    def delete(self, name: str, **kwargs: t.Any) -> None:
2887        cascade = kwargs.pop("cascade", False)
2888        try:
2889            # Some engines (e.g., RisingWave) don’t fail when dropping a materialized view with a DROP VIEW statement,
2890            # because views and materialized views don’t share the same namespace. Therefore, we should not ignore if the
2891            # view doesn't exist and let the exception handler attempt to drop the materialized view.
2892            self.adapter.drop_view(name, cascade=cascade, ignore_if_not_exists=False)
2893        except Exception:
2894            logger.debug(
2895                "Failed to drop view '%s'. Trying to drop the materialized view instead",
2896                name,
2897                exc_info=True,
2898            )
2899            self.adapter.drop_view(
2900                name, materialized=True, cascade=cascade, ignore_if_not_exists=True
2901            )
2902        logger.info("Dropped view '%s'", name)
2903
2904    def _is_materialized_view(self, model: Model) -> bool:
2905        return isinstance(model.kind, ViewKind) and model.kind.materialized
2906
2907
2908C = t.TypeVar("C", bound=CustomKind)
2909
2910
2911class CustomMaterialization(IncrementalStrategy, t.Generic[C]):
2912    """Base class for custom materializations."""
2913
2914    def insert(
2915        self,
2916        table_name: str,
2917        query_or_df: QueryOrDF,
2918        model: Model,
2919        is_first_insert: bool,
2920        render_kwargs: t.Dict[str, t.Any],
2921        **kwargs: t.Any,
2922    ) -> None:
2923        """Inserts the given query or a DataFrame into the target table or a view.
2924
2925        Args:
2926            table_name: The name of the target table or view.
2927            query_or_df: A query or a DataFrame to insert.
2928            model: The target model.
2929            is_first_insert: Whether this is the first insert for this version of a model. This value is set to True
2930                if no data has been previously inserted into the target table, or when the entire history of the target model has
2931                been restated. Note that in the latter case, the table might contain data from previous executions, and it is the
2932                responsibility of a specific evaluation strategy to handle the truncation of the table if necessary.
2933            render_kwargs: Additional key-value arguments to pass when rendering the model's query.
2934        """
2935        raise NotImplementedError(
2936            "Custom materialization strategies must implement the 'insert' method."
2937        )
2938
2939
2940_custom_materialization_type_cache: t.Optional[
2941    t.Dict[str, t.Tuple[t.Type[CustomKind], t.Type[CustomMaterialization]]]
2942] = None
2943
2944
2945def get_custom_materialization_kind_type(st: t.Type[CustomMaterialization]) -> t.Type[CustomKind]:
2946    # try to read if there is a custom 'kind' type in use by inspecting the type signature
2947    # eg try to read 'MyCustomKind' from:
2948    # >>>> class MyCustomMaterialization(CustomMaterialization[MyCustomKind])
2949    # and fall back to base CustomKind if there is no generic type declared
2950    if hasattr(st, "__orig_bases__"):
2951        for base in st.__orig_bases__:
2952            if hasattr(base, "__origin__") and base.__origin__ == CustomMaterialization:
2953                for generic_arg in t.get_args(base):
2954                    if not issubclass(generic_arg, CustomKind):
2955                        raise SQLMeshError(
2956                            f"Custom materialization kind '{generic_arg.__name__}' must be a subclass of CustomKind"
2957                        )
2958
2959                    return generic_arg
2960
2961    return CustomKind
2962
2963
2964def get_custom_materialization_type(
2965    name: str, raise_errors: bool = True
2966) -> t.Optional[t.Tuple[t.Type[CustomKind], t.Type[CustomMaterialization]]]:
2967    global _custom_materialization_type_cache
2968
2969    strategy_key = name.lower()
2970
2971    try:
2972        if (
2973            _custom_materialization_type_cache is None
2974            or strategy_key not in _custom_materialization_type_cache
2975        ):
2976            strategy_types = list(CustomMaterialization.__subclasses__())
2977
2978            entry_points = metadata.entry_points(group="sqlmesh.materializations")
2979            for entry_point in entry_points:
2980                strategy_type = entry_point.load()
2981                if not issubclass(strategy_type, CustomMaterialization):
2982                    raise SQLMeshError(
2983                        f"Custom materialization entry point '{entry_point.name}' must be a subclass of CustomMaterialization."
2984                    )
2985                strategy_types.append(strategy_type)
2986
2987            _custom_materialization_type_cache = {
2988                getattr(strategy_type, "NAME", strategy_type.__name__).lower(): (
2989                    get_custom_materialization_kind_type(strategy_type),
2990                    strategy_type,
2991                )
2992                for strategy_type in strategy_types
2993            }
2994
2995        if strategy_key not in _custom_materialization_type_cache:
2996            raise ConfigError(f"Materialization strategy with name '{name}' was not found.")
2997    except (SQLMeshError, ConfigError) as e:
2998        if raise_errors:
2999            raise e
3000
3001        from sqlmesh.core.console import get_console
3002
3003        get_console().log_warning(str(e))
3004        return None
3005
3006    strategy_kind_type, strategy_type = _custom_materialization_type_cache[strategy_key]
3007    logger.debug(
3008        "Resolved custom materialization '%s' to '%s' (%s)", name, strategy_type, strategy_kind_type
3009    )
3010
3011    return strategy_kind_type, strategy_type
3012
3013
3014def get_custom_materialization_type_or_raise(
3015    name: str,
3016) -> t.Tuple[t.Type[CustomKind], t.Type[CustomMaterialization]]:
3017    types = get_custom_materialization_type(name, raise_errors=True)
3018    if types is not None:
3019        return types[0], types[1]
3020
3021    # Shouldnt get here as get_custom_materialization_type() has raise_errors=True, but just in case...
3022    raise SQLMeshError(f"Custom materialization '{name}' not present in the Python environment")
3023
3024
3025class DbtCustomMaterializationStrategy(MaterializableStrategy):
3026    def __init__(
3027        self,
3028        adapter: EngineAdapter,
3029        materialization_name: str,
3030        materialization_template: str,
3031    ):
3032        super().__init__(adapter)
3033        self.materialization_name = materialization_name
3034        self.materialization_template = materialization_template
3035
3036    def create(
3037        self,
3038        table_name: str,
3039        model: Model,
3040        is_table_deployable: bool,
3041        render_kwargs: t.Dict[str, t.Any],
3042        skip_grants: bool,
3043        **kwargs: t.Any,
3044    ) -> None:
3045        original_query = model.render_query_or_raise(**render_kwargs)
3046        self._execute_materialization(
3047            table_name=table_name,
3048            query_or_df=original_query.limit(0),
3049            model=model,
3050            is_first_insert=True,
3051            render_kwargs=render_kwargs,
3052            create_only=True,
3053            **kwargs,
3054        )
3055
3056        # Apply grants after dbt custom materialization table creation
3057        if not skip_grants:
3058            is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
3059            self._apply_grants(
3060                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
3061            )
3062
3063    def insert(
3064        self,
3065        table_name: str,
3066        query_or_df: QueryOrDF,
3067        model: Model,
3068        is_first_insert: bool,
3069        render_kwargs: t.Dict[str, t.Any],
3070        **kwargs: t.Any,
3071    ) -> None:
3072        self._execute_materialization(
3073            table_name=table_name,
3074            query_or_df=query_or_df,
3075            model=model,
3076            is_first_insert=is_first_insert,
3077            render_kwargs=render_kwargs,
3078            **kwargs,
3079        )
3080
3081        # Apply grants after custom materialization insert (only on first insert)
3082        if is_first_insert:
3083            is_snapshot_deployable = kwargs.get("is_snapshot_deployable", False)
3084            self._apply_grants(
3085                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
3086            )
3087
3088    def append(
3089        self,
3090        table_name: str,
3091        query_or_df: QueryOrDF,
3092        model: Model,
3093        render_kwargs: t.Dict[str, t.Any],
3094        **kwargs: t.Any,
3095    ) -> None:
3096        return self.insert(
3097            table_name,
3098            query_or_df,
3099            model,
3100            is_first_insert=False,
3101            render_kwargs=render_kwargs,
3102            **kwargs,
3103        )
3104
3105    def run_pre_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
3106        # in dbt custom materialisations it's up to the user to run the pre hooks inside the transaction
3107        if not render_kwargs.get("inside_transaction", True):
3108            super().run_pre_statements(
3109                snapshot=snapshot,
3110                render_kwargs=render_kwargs,
3111            )
3112
3113    def run_post_statements(self, snapshot: Snapshot, render_kwargs: t.Any) -> None:
3114        # in dbt custom materialisations it's up to the user to run the post hooks inside the transaction
3115        if not render_kwargs.get("inside_transaction", True):
3116            super().run_post_statements(
3117                snapshot=snapshot,
3118                render_kwargs=render_kwargs,
3119            )
3120
3121    def _execute_materialization(
3122        self,
3123        table_name: str,
3124        query_or_df: QueryOrDF,
3125        model: Model,
3126        is_first_insert: bool,
3127        render_kwargs: t.Dict[str, t.Any],
3128        create_only: bool = False,
3129        **kwargs: t.Any,
3130    ) -> None:
3131        jinja_macros = model.jinja_macros
3132
3133        # For vdes we need to use the table, since we don't know the schema/table at parse time
3134        parts = exp.to_table(table_name, dialect=self.adapter.dialect)
3135
3136        existing_globals = jinja_macros.global_objs
3137        relation_info = existing_globals.get("this")
3138        if isinstance(relation_info, dict):
3139            relation_info["database"] = parts.catalog
3140            relation_info["identifier"] = parts.name
3141            relation_info["name"] = parts.name
3142
3143        jinja_globals = {
3144            **existing_globals,
3145            "this": relation_info,
3146            "database": parts.catalog,
3147            "schema": parts.db,
3148            "identifier": parts.name,
3149            "target": existing_globals.get("target", {"type": self.adapter.dialect}),
3150            "execution_dt": kwargs.get("execution_time"),
3151            "engine_adapter": self.adapter,
3152            "sql": str(query_or_df),
3153            "is_first_insert": is_first_insert,
3154            "create_only": create_only,
3155            "pre_hooks": [
3156                AttributeDict({"sql": s.this.this, "transaction": transaction})
3157                for s in model.pre_statements
3158                if (transaction := s.args.get("transaction", True))
3159            ],
3160            "post_hooks": [
3161                AttributeDict({"sql": s.this.this, "transaction": transaction})
3162                for s in model.post_statements
3163                if (transaction := s.args.get("transaction", True))
3164            ],
3165            "model_instance": model,
3166            **kwargs,
3167        }
3168
3169        try:
3170            jinja_env = jinja_macros.build_environment(**jinja_globals)
3171            template = jinja_env.from_string(self.materialization_template)
3172
3173            try:
3174                template.render()
3175            except MacroReturnVal as ret:
3176                # this is a successful return from a macro call (dbt uses this list of Relations to update their relation cache)
3177                returned_relations = ret.value.get("relations", [])
3178                logger.info(
3179                    f"Materialization {self.materialization_name} returned relations: {returned_relations}"
3180                )
3181
3182        except Exception as e:
3183            raise SQLMeshError(
3184                f"Failed to execute dbt materialization '{self.materialization_name}': {e}"
3185            ) from e
3186
3187
3188class EngineManagedStrategy(MaterializableStrategy):
3189    def create(
3190        self,
3191        table_name: str,
3192        model: Model,
3193        is_table_deployable: bool,
3194        render_kwargs: t.Dict[str, t.Any],
3195        skip_grants: bool,
3196        **kwargs: t.Any,
3197    ) -> None:
3198        is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
3199
3200        if is_table_deployable and is_snapshot_deployable:
3201            # We could deploy this to prod; create a proper managed table
3202            logger.info("Creating managed table: %s", table_name)
3203            physical_properties = _adjust_physical_properties_for_engine(
3204                self.adapter, model, kwargs.get("physical_properties", model.physical_properties)
3205            )
3206            self.adapter.create_managed_table(
3207                table_name=table_name,
3208                query=model.render_query_or_raise(**render_kwargs),
3209                target_columns_to_types=model.columns_to_types,
3210                partitioned_by=model.partitioned_by,
3211                clustered_by=model.clustered_by,  # type: ignore[arg-type]
3212                table_properties=physical_properties,
3213                table_description=model.description,
3214                column_descriptions=model.column_descriptions,
3215                table_format=model.table_format,
3216            )
3217
3218            # Apply grants after managed table creation
3219            if not skip_grants:
3220                self._apply_grants(
3221                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
3222                )
3223
3224        elif not is_table_deployable:
3225            # Only create the dev preview table as a normal table.
3226            # For the main table, if the snapshot is cant be deployed to prod (eg upstream is forward-only) do nothing.
3227            # Any downstream models that reference it will be updated to point to the dev preview table.
3228            # If the user eventually tries to deploy it, the logic in insert() will see it doesnt exist and create it
3229            super().create(
3230                table_name=table_name,
3231                model=model,
3232                is_table_deployable=is_table_deployable,
3233                render_kwargs=render_kwargs,
3234                skip_grants=skip_grants,
3235                **kwargs,
3236            )
3237
3238    def insert(
3239        self,
3240        table_name: str,
3241        query_or_df: QueryOrDF,
3242        model: Model,
3243        is_first_insert: bool,
3244        render_kwargs: t.Dict[str, t.Any],
3245        **kwargs: t.Any,
3246    ) -> None:
3247        deployability_index: DeployabilityIndex = kwargs["deployability_index"]
3248        snapshot: Snapshot = kwargs["snapshot"]
3249        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
3250        if is_first_insert and is_snapshot_deployable and not self.adapter.table_exists(table_name):
3251            self.adapter.create_managed_table(
3252                table_name=table_name,
3253                query=query_or_df,  # type: ignore
3254                target_columns_to_types=model.columns_to_types,
3255                partitioned_by=model.partitioned_by,
3256                clustered_by=model.clustered_by,  # type: ignore[arg-type]
3257                table_properties=kwargs.get("physical_properties", model.physical_properties),
3258                table_description=model.description,
3259                column_descriptions=model.column_descriptions,
3260                table_format=model.table_format,
3261            )
3262            self._apply_grants(
3263                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
3264            )
3265        elif not is_snapshot_deployable:
3266            # Snapshot isnt deployable; update the preview table instead
3267            # If the snapshot was deployable, then data would have already been loaded in create() because a managed table would have been created
3268            logger.info(
3269                "Updating preview table: %s (for managed model: %s)",
3270                table_name,
3271                model.name,
3272            )
3273            self._replace_query_for_model(
3274                model=model,
3275                name=table_name,
3276                query_or_df=query_or_df,
3277                render_kwargs=render_kwargs,
3278                **kwargs,
3279            )
3280
3281    def append(
3282        self,
3283        table_name: str,
3284        query_or_df: QueryOrDF,
3285        model: Model,
3286        render_kwargs: t.Dict[str, t.Any],
3287        **kwargs: t.Any,
3288    ) -> None:
3289        raise ConfigError(f"Cannot append to a managed table '{table_name}'.")
3290
3291    def migrate(
3292        self,
3293        target_table_name: str,
3294        source_table_name: str,
3295        snapshot: Snapshot,
3296        *,
3297        ignore_destructive: bool,
3298        ignore_additive: bool,
3299        **kwargs: t.Any,
3300    ) -> None:
3301        potential_alter_operations = self.adapter.get_alter_operations(
3302            target_table_name,
3303            source_table_name,
3304            ignore_destructive=ignore_destructive,
3305            ignore_additive=ignore_additive,
3306        )
3307        if len(potential_alter_operations) > 0:
3308            # this can happen if a user changes a managed model and deliberately overrides a plan to be forward only, eg `sqlmesh plan --forward-only`
3309            raise MigrationNotSupportedError(
3310                f"The schema of the managed model '{target_table_name}' cannot be updated in a forward-only fashion."
3311            )
3312
3313        # Apply grants after verifying no schema changes
3314        deployability_index = kwargs.get("deployability_index")
3315        is_snapshot_deployable = (
3316            deployability_index.is_deployable(snapshot) if deployability_index else False
3317        )
3318        self._apply_grants(
3319            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
3320        )
3321
3322    def delete(self, name: str, **kwargs: t.Any) -> None:
3323        # a dev preview table is created as a normal table, so it needs to be dropped as a normal table
3324        _check_table_db_is_physical_schema(name, kwargs["physical_schema"])
3325        if kwargs["is_table_deployable"]:
3326            self.adapter.drop_managed_table(name)
3327            logger.info("Dropped managed table '%s'", name)
3328        else:
3329            self.adapter.drop_table(name)
3330            logger.info("Dropped dev preview for managed table '%s'", name)
3331
3332
3333def _intervals(snapshot: Snapshot, deployability_index: DeployabilityIndex) -> Intervals:
3334    return (
3335        snapshot.intervals
3336        if deployability_index.is_deployable(snapshot)
3337        else snapshot.dev_intervals
3338    )
3339
3340
3341def _check_destructive_schema_change(
3342    snapshot: Snapshot,
3343    alter_operations: t.List[TableAlterOperation],
3344    allow_destructive_snapshots: t.Set[str],
3345) -> None:
3346    if (
3347        snapshot.is_no_rebuild
3348        and snapshot.needs_destructive_check(allow_destructive_snapshots)
3349        and has_drop_alteration(alter_operations)
3350    ):
3351        snapshot_name = snapshot.name
3352        model_dialect = snapshot.model.dialect
3353
3354        if snapshot.model.on_destructive_change.is_warn:
3355            logger.warning(
3356                format_destructive_change_msg(
3357                    snapshot_name,
3358                    alter_operations,
3359                    model_dialect,
3360                    error=False,
3361                )
3362            )
3363            return
3364        raise DestructiveChangeError(
3365            format_destructive_change_msg(snapshot_name, alter_operations, model_dialect)
3366        )
3367
3368
3369def _check_additive_schema_change(
3370    snapshot: Snapshot,
3371    alter_operations: t.List[TableAlterOperation],
3372    allow_additive_snapshots: t.Set[str],
3373) -> None:
3374    # Only check additive changes for incremental models that have the on_additive_change property
3375    if not isinstance(snapshot.model.kind, _Incremental):
3376        return
3377
3378    if snapshot.needs_additive_check(allow_additive_snapshots) and has_additive_alteration(
3379        alter_operations
3380    ):
3381        # Note: IGNORE filtering is applied before this function is called
3382        # so if we reach here, additive changes are not being ignored
3383        snapshot_name = snapshot.name
3384        model_dialect = snapshot.model.dialect
3385
3386        if snapshot.model.on_additive_change.is_warn:
3387            logger.warning(
3388                format_additive_change_msg(
3389                    snapshot_name,
3390                    alter_operations,
3391                    model_dialect,
3392                    error=False,
3393                )
3394            )
3395            return
3396        if snapshot.model.on_additive_change.is_error:
3397            raise AdditiveChangeError(
3398                format_additive_change_msg(snapshot_name, alter_operations, model_dialect)
3399            )
3400
3401
3402def _check_table_db_is_physical_schema(table_name: str, physical_schema: str) -> None:
3403    table = exp.to_table(table_name)
3404    if table.db != physical_schema:
3405        raise SQLMeshError(
3406            f"Table '{table_name}' is not a part of the physical schema '{physical_schema}' and so can't be dropped."
3407        )
3408
3409
3410def _snapshot_to_data_object_type(snapshot: Snapshot) -> DataObjectType:
3411    if snapshot.is_managed:
3412        return DataObjectType.MANAGED_TABLE
3413    if snapshot.is_materialized_view:
3414        return DataObjectType.MATERIALIZED_VIEW
3415    if snapshot.is_view:
3416        return DataObjectType.VIEW
3417    if snapshot.is_materialized:
3418        return DataObjectType.TABLE
3419    return DataObjectType.UNKNOWN