Feat: Add support for the non-virtual prod mode

Author: izeigerman

sqlmesh/core/config/common.py

@@ -49,6 +49,35 @@ def __repr__(self) -> str:
         return str(self)
 
 
+class VirtualEnvironmentMode(str, Enum):
+    """Mode for virtual environment behavior.
+
+    FULL: Use full virtual environment functionality with hashed table names and virtual layer updates.
+    DEV_ONLY: Bypass virtual environments in production, using simple table names without hashes.
+    """
+
+    FULL = "full"
+    DEV_ONLY = "dev_only"
+
+    @property
+    def is_full(self) -> bool:
+        return self == VirtualEnvironmentMode.FULL
+
+    @property
+    def is_dev_only(self) -> bool:
+        return self == VirtualEnvironmentMode.DEV_ONLY
+
+    @classproperty
+    def default(cls) -> VirtualEnvironmentMode:
+        return VirtualEnvironmentMode.FULL
+
+    def __str__(self) -> str:
+        return self.name
+
+    def __repr__(self) -> str:
+        return str(self)
+
+
 class TableNamingConvention(str, Enum):
     # Causes table names at the physical layer to follow the convention:
     # <schema-name>__<table-name>__<fingerprint>

sqlmesh/core/config/root.py

@@ -14,7 +14,11 @@
 from sqlmesh.cicd.config import CICDBotConfig
 from sqlmesh.core import constants as c
 from sqlmesh.core.console import get_console
-from sqlmesh.core.config import EnvironmentSuffixTarget, TableNamingConvention
+from sqlmesh.core.config.common import (
+    EnvironmentSuffixTarget,
+    TableNamingConvention,
+    VirtualEnvironmentMode,
+)
 from sqlmesh.core.config.base import BaseConfig, UpdateStrategy
 from sqlmesh.core.config.common import variables_validator, compile_regex_mapping
 from sqlmesh.core.config.connection import (
@@ -107,6 +111,7 @@ class Config(BaseConfig):
         physical_schema_mapping: A mapping from regular expressions to names of schemas in which physical tables for corresponding models will be placed.
         environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
         physical_table_naming_convention: Indicates how tables should be named at the physical layer
+        virtual_environment_mode: Indicates how environments should be handled.
         gateway_managed_virtual_layer: Whether the models' views in the virtual layer are created by the model-specific gateway rather than the default gateway.
         infer_python_dependencies: Whether to statically analyze Python code to automatically infer Python package requirements.
         environment_catalog_mapping: A mapping from regular expressions to catalog names. The catalog name is used to determine the target catalog for a given environment.
@@ -148,9 +153,8 @@ class Config(BaseConfig):
     environment_suffix_target: EnvironmentSuffixTarget = Field(
         default=EnvironmentSuffixTarget.default
     )
-    physical_table_naming_convention: TableNamingConvention = Field(
-        default=TableNamingConvention.default
-    )
+    physical_table_naming_convention: TableNamingConvention = TableNamingConvention.default
+    virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default
     gateway_managed_virtual_layer: bool = False
     infer_python_dependencies: bool = True
     environment_catalog_mapping: RegexKeyDict = {}

sqlmesh/core/context.py

@@ -2909,6 +2909,7 @@ def _nodes_to_snapshots(self, nodes: t.Dict[str, Node]) -> t.Dict[str, Snapshot]
                 config = self.config_for_node(node)
                 kwargs["ttl"] = config.snapshot_ttl
                 kwargs["table_naming_convention"] = config.physical_table_naming_convention
+                kwargs["virtual_environment_mode"] = config.virtual_environment_mode
 
             snapshot = Snapshot.from_node(
                 node,
@@ -2936,7 +2937,7 @@ def _node_or_snapshot_to_fqn(self, node_or_snapshot: NodeOrSnapshot) -> str:
     def _plan_preview_enabled(self) -> bool:
         if self.config.plan.enable_preview is not None:
             return self.config.plan.enable_preview
-        # It is dangerous to enable preview by default for dbt projects that rely on engines that don’t support cloning.
+        # It is dangerous to enable preview by default for dbt projects that rely on engines that don't support cloning.
         # Enabling previews in such cases can result in unintended full refreshes because dbt incremental models rely on
         # the maximum timestamp value in the target table.
         return self._project_type == c.NATIVE or self.engine_adapter.SUPPORTS_CLONING

sqlmesh/core/plan/builder.py

@@ -811,7 +811,7 @@ def _ensure_no_forward_only_revert(self) -> None:
                 and not candidate.model.forward_only
                 and promoted.is_forward_only
                 and not promoted.is_paused
-                and not candidate.reuses_previous_version
+                and not candidate.is_no_rebuild
                 and promoted.version == candidate.version
             ):
                 raise PlanError(

sqlmesh/core/plan/stages.py

@@ -370,6 +370,7 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
             demoted_environment_naming_info,
             snapshots | full_demoted_snapshots,
             deployability_index,
+            plan.is_dev,
         )
         if virtual_layer_update_stage:
             stages.append(virtual_layer_update_stage)
@@ -449,11 +450,18 @@ def _get_virtual_layer_update_stage(
         demoted_environment_naming_info: t.Optional[EnvironmentNamingInfo],
         all_snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: DeployabilityIndex,
+        is_dev: bool,
     ) -> t.Optional[VirtualLayerUpdateStage]:
-        promoted_snapshots = {s for s in promoted_snapshots if s.is_model and not s.is_symbolic}
-        demoted_snapshots = {s for s in demoted_snapshots if s.is_model and not s.is_symbolic}
+        def _should_update_virtual_layer(snapshot: SnapshotTableInfo) -> bool:
+            # Skip virtual layer update for snapshots with virtual environment support disabled
+            virtual_environment_enabled = is_dev or snapshot.virtual_environment_mode.is_full
+            return snapshot.is_model and not snapshot.is_symbolic and virtual_environment_enabled
+
+        promoted_snapshots = {s for s in promoted_snapshots if _should_update_virtual_layer(s)}
+        demoted_snapshots = {s for s in demoted_snapshots if _should_update_virtual_layer(s)}
         if not promoted_snapshots and not demoted_snapshots:
             return None
+
         return VirtualLayerUpdateStage(
             promoted_snapshots=promoted_snapshots,
             demoted_snapshots=demoted_snapshots,

sqlmesh/core/snapshot/definition.py

@@ -13,7 +13,7 @@
 from sqlglot import exp
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 
-from sqlmesh.core.config import TableNamingConvention
+from sqlmesh.core.config.common import TableNamingConvention, VirtualEnvironmentMode
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import StandaloneAudit
 from sqlmesh.core.environment import EnvironmentSuffixTarget
@@ -229,6 +229,7 @@ class SnapshotDataVersion(PydanticModel, frozen=True):
     physical_schema_: t.Optional[str] = Field(default=None, alias="physical_schema")
     dev_table_suffix: str
     table_naming_convention: TableNamingConvention = Field(default=TableNamingConvention.default)
+    virtual_environment_mode: VirtualEnvironmentMode = Field(default=VirtualEnvironmentMode.default)
 
     def snapshot_id(self, name: str) -> SnapshotId:
         return SnapshotId(name=name, identifier=self.fingerprint.to_identifier())
@@ -335,7 +336,8 @@ class SnapshotInfoMixin(ModelKindMixin):
     # This can be removed from this model once Pydantic 1 support is dropped (must remain in `Snapshot` though)
     base_table_name_override: t.Optional[str]
     dev_table_suffix: str
-    table_naming_convention: TableNamingConvention = Field(default=TableNamingConvention.default)
+    table_naming_convention: TableNamingConvention
+    virtual_environment_mode: VirtualEnvironmentMode
 
     @cached_property
     def identifier(self) -> str:
@@ -394,7 +396,7 @@ def is_indirect_non_breaking(self) -> bool:
         return self.change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING
 
     @property
-    def reuses_previous_version(self) -> bool:
+    def is_no_rebuild(self) -> bool:
         return self.change_category in (
             SnapshotChangeCategory.FORWARD_ONLY,
             SnapshotChangeCategory.METADATA,
@@ -432,6 +434,10 @@ def _table_name(self, version: str, is_deployable: bool) -> str:
         if self.is_external:
             return self.name
 
+        if is_deployable and self.virtual_environment_mode.is_dev_only:
+            # Use the model name as is if the target is deployable and the virtual environment mode is set to dev-only
+            return self.name
+
         is_dev_table = not is_deployable
         if is_dev_table:
             version = self.dev_version
@@ -448,6 +454,7 @@ def _table_name(self, version: str, is_deployable: bool) -> str:
             fqt = self.fully_qualified_table.copy()
             fqt.set("catalog", None)
             base_table_name = fqt.sql()
+
         return table_name(
             self.physical_schema,
             base_table_name,
@@ -487,6 +494,8 @@ class SnapshotTableInfo(PydanticModel, SnapshotInfoMixin, frozen=True):
     custom_materialization: t.Optional[str] = None
     dev_table_suffix: str
     model_gateway: t.Optional[str] = None
+    table_naming_convention: TableNamingConvention = Field(default=TableNamingConvention.default)
+    virtual_environment_mode: VirtualEnvironmentMode = Field(default=VirtualEnvironmentMode.default)
 
     def __lt__(self, other: SnapshotTableInfo) -> bool:
         return self.name < other.name
@@ -528,6 +537,7 @@ def data_version(self) -> SnapshotDataVersion:
             physical_schema=self.physical_schema,
             dev_table_suffix=self.dev_table_suffix,
             table_naming_convention=self.table_naming_convention,
+            virtual_environment_mode=self.virtual_environment_mode,
         )
 
     @property
@@ -614,6 +624,7 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     table_naming_convention_: TableNamingConvention = Field(
         default=TableNamingConvention.default, alias="table_naming_convention"
     )
+    virtual_environment_mode: VirtualEnvironmentMode = Field(default=VirtualEnvironmentMode.default)
 
     @field_validator("ttl")
     @classmethod
@@ -666,6 +677,7 @@ def from_node(
         version: t.Optional[str] = None,
         cache: t.Optional[t.Dict[str, SnapshotFingerprint]] = None,
         table_naming_convention: TableNamingConvention = TableNamingConvention.default,
+        virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default,
     ) -> Snapshot:
         """Creates a new snapshot for a node.
 
@@ -677,6 +689,7 @@ def from_node(
             version: The version that a snapshot is associated with. Usually set during the planning phase.
             cache: Cache of node name to fingerprints.
             table_naming_convention: Convention to follow when generating the physical table name
+            virtual_environment_mode: Mode for handling virtual environments
 
         Returns:
             The newly created snapshot.
@@ -709,6 +722,7 @@ def from_node(
             ttl=ttl,
             version=version,
             table_naming_convention=table_naming_convention,
+            virtual_environment_mode=virtual_environment_mode,
         )
 
     def __eq__(self, other: t.Any) -> bool:
@@ -863,16 +877,19 @@ def merge_intervals(self, other: t.Union[Snapshot, SnapshotIntervals]) -> None:
         Args:
             other: The target snapshot to inherit intervals from.
         """
-        effective_from_ts = self.normalized_effective_from_ts or 0
-        apply_effective_from = effective_from_ts > 0 and self.identifier != other.identifier
-
-        for start, end in other.intervals:
-            # If the effective_from is set, then intervals that come after it must come from
-            # the current snapshost.
-            if apply_effective_from and start < effective_from_ts:
-                end = min(end, effective_from_ts)
-            if not apply_effective_from or end <= effective_from_ts:
-                self.add_interval(start, end)
+        if self.is_no_rebuild or self.virtual_environment_mode.is_full or not self.is_paused:
+            # If the virtual environment mode is not full we can only merge prod intervals if this snapshot
+            # is currently promoted in production or if it's forward-only / metadata / indirect non-breaking.
+            # Otherwise, we want to ignore any existing intervals and backfill this snapshot from scratch.
+            effective_from_ts = self.normalized_effective_from_ts or 0
+            apply_effective_from = effective_from_ts > 0 and self.identifier != other.identifier
+            for start, end in other.intervals:
+                # If the effective_from is set, then intervals that come after it must come from
+                # the current snapshost.
+                if apply_effective_from and start < effective_from_ts:
+                    end = min(end, effective_from_ts)
+                if not apply_effective_from or end <= effective_from_ts:
+                    self.add_interval(start, end)
 
         if self.dev_version == other.dev_version:
             # Merge dev intervals if the dev versions match which would mean
@@ -1013,15 +1030,18 @@ def categorize_as(self, category: SnapshotChangeCategory) -> None:
             category: The change category to assign to this snapshot.
         """
         self.dev_version_ = self.fingerprint.to_version()
-        reuse_previous_version = category in (
+        is_no_rebuild = category in (
             SnapshotChangeCategory.FORWARD_ONLY,
             SnapshotChangeCategory.INDIRECT_NON_BREAKING,
             SnapshotChangeCategory.METADATA,
         )
-        if self.is_model and self.model.physical_version:
+        if self.is_model and not self.virtual_environment_mode.is_full:
+            # Hardcode the version if the virtual environment is not fully enabled.
+            self.version = "novde"
+        elif self.is_model and self.model.physical_version:
             # If the model has a pinned version then use that.
             self.version = self.model.physical_version
-        elif reuse_previous_version and self.previous_version:
+        elif is_no_rebuild and self.previous_version:
             previous_version = self.previous_version
             self.version = previous_version.data_version.version
             self.physical_schema_ = previous_version.physical_schema
@@ -1219,7 +1239,8 @@ def table_info(self) -> SnapshotTableInfo:
             custom_materialization=custom_materialization,
             dev_table_suffix=self.dev_table_suffix,
             model_gateway=self.model_gateway,
-            table_naming_convention=self.table_naming_convention,  # type: ignore
+            table_naming_convention=self.table_naming_convention,
+            virtual_environment_mode=self.virtual_environment_mode,
         )
 
     @property
@@ -1233,6 +1254,7 @@ def data_version(self) -> SnapshotDataVersion:
             physical_schema=self.physical_schema,
             dev_table_suffix=self.dev_table_suffix,
             table_naming_convention=self.table_naming_convention,
+            virtual_environment_mode=self.virtual_environment_mode,
         )
 
     @property
@@ -1501,14 +1523,20 @@ def create(
         for node in dag:
             if node not in snapshots:
                 continue
-            # Make sure that the node is deployable according to all its parents
-            this_deployable = all(
-                children_deployability_mapping[p_id]
-                for p_id in snapshots[node].parents
-                if p_id in children_deployability_mapping
-            )
+            snapshot = snapshots[node]
+
+            if not snapshot.virtual_environment_mode.is_full:
+                # If the virtual environment is not fully enabled, then the snapshot can never be deployable
+                this_deployable = False
+            else:
+                # Make sure that the node is deployable according to all its parents
+                this_deployable = all(
+                    children_deployability_mapping[p_id]
+                    for p_id in snapshots[node].parents
+                    if p_id in children_deployability_mapping
+                )
+
             if this_deployable:
-                snapshot = snapshots[node]
                 is_forward_only_model = (
                     snapshot.is_model and snapshot.model.forward_only and not snapshot.is_metadata
                 )

sqlmesh/core/snapshot/evaluator.py

@@ -338,7 +338,7 @@ def create(
                 continue
             deployability_flags = [True]
             if (
-                snapshot.reuses_previous_version
+                snapshot.is_no_rebuild
                 or snapshot.is_managed
                 or (snapshot.is_model and snapshot.model.forward_only)
                 or (deployability_index and not deployability_index.is_deployable(snapshot))
@@ -2278,8 +2278,10 @@ def _check_destructive_schema_change(
     alter_expressions: t.List[exp.Alter],
     allow_destructive_snapshots: t.Set[str],
 ) -> None:
-    if snapshot.needs_destructive_check(allow_destructive_snapshots) and has_drop_alteration(
-        alter_expressions
+    if (
+        snapshot.is_no_rebuild
+        and snapshot.needs_destructive_check(allow_destructive_snapshots)
+        and has_drop_alteration(alter_expressions)
     ):
         snapshot_name = snapshot.name
         dropped_column_names = get_dropped_column_names(alter_expressions)

tests/core/test_snapshot.py

@@ -167,6 +167,7 @@ def test_json(snapshot: Snapshot):
         "parents": [{"name": '"parent"."tbl"', "identifier": snapshot.parents[0].identifier}],
         "previous_versions": [],
         "table_naming_convention": "schema_and_table",
+        "virtual_environment_mode": "full",
         "updated_ts": 1663891973000,
         "version": snapshot.fingerprint.to_version(),
         "migrated": False,