Feat!: Dev-only VDE mode (#5087)

Author: izeigerman

docs/guides/configuration.md

@@ -538,6 +538,44 @@ sqlmesh_md5__d3b07384d113edec49eaa6238ad5ff00__dev
 
 This has a downside that now it's much more difficult to determine which table corresponds to which model by just looking at the database with a SQL client. However, the table names have a predictable length so there are no longer any surprises with identfiers exceeding the max length at the physical layer.
 
+#### Virtual Data Environment Modes
+
+By default, Virtual Data Environments (VDE) are applied across both development and production environments. This allows SQLMesh to reuse physical tables when appropriate, even when promoting from development to production.
+
+However, users may prefer their production environment to be non-virtual. The non-exhaustive list of reasons may include:
+
+- Integration with third-party tools and platforms, such as data catalogs, may not work well with the virtual view layer that SQLMesh imposes by default
+- A desire to rely on time travel features provided by cloud data warehouses such as BigQuery, Snowflake, and Databricks
+
+To mitigate this, SQLMesh offers an alternative 'dev-only' mode for using VDE. It can be enabled in the project configuration like so:
+
+=== "YAML"
+
+    ```yaml linenums="1"
+    virtual_environment_mode: dev_only
+    ```
+
+=== "Python"
+
+    ```python linenums="1"
+    from sqlmesh.core.config import Config
+
+    config = Config(
+        virtual_environment_mode="dev_only",
+    )
+    ```
+
+'dev-only' mode means that VDE is applied only in development environments. While in production, model tables and views are updated directly and bypass the virtual layer. This also means that physical tables in production will be created using the original, **unversioned** model names. Users will still benefit from VDE and data reuse across development environments.
+
+Please note the following tradeoffs when enabling this mode:
+
+- All data inserted in development environments is used only for [preview](../concepts/plans.md#data-preview-for-forward-only-changes) and will **not** be reused in production
+- Reverting a model to a previous version will be applied going forward and may require an explicit data restatement
+
+!!! warning
+    Switching the mode for an existing project will result in a **complete rebuild** of all models in the project. Refer to the [Table Migration Guide](./table_migration.md) to migrate existing tables without rebuilding them from scratch.
+
+
 #### Environment view catalogs
 
 By default, SQLMesh creates an environment view in the same [catalog](../concepts/glossary.md#catalog) as the physical table the view points to. The physical table's catalog is determined by either the catalog specified in the model name or the default catalog defined in the connection.

docs/reference/configuration.md

@@ -46,6 +46,7 @@ Configuration options for how SQLMesh manages environment creation and promotion
 | `environment_suffix_target`   | Whether SQLMesh views should append their environment name to the `schema`, `table` or `catalog` - [additional details](../guides/configuration.md#view-schema-override). (Default: `schema`)                                                                                                      | string               | N        |
 | `gateway_managed_virtual_layer`   | Whether SQLMesh views of the virtual layer will be created by the default gateway or model specified gateways - [additional details](../guides/multi_engine.md#gateway-managed-virtual-layer). (Default: False)                                                                                | boolean              | N        |
 | `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.                                                                                                                                                             | dict[string, string] | N        |
+| `virtual_environment_mode` | Determines the Virtual Data Environment (VDE) mode. If set to `full`, VDE is used in both production and development environments. The `dev_only` option enables VDE only in development environments, while in production, no virtual layer is used and models are materialized directly using their original names (i.e., no versioned physical tables). (Default: `full`) | string | N |
 
 ### Models
 

examples/sushi/config.py

@@ -1,5 +1,6 @@
 import os
 
+from sqlmesh.core.config.common import VirtualEnvironmentMode
 from sqlmesh.core.config import (
     AutoCategorizationMode,
     BigQueryConnectionConfig,
@@ -76,6 +77,16 @@
     model_defaults=model_defaults,
 )
 
+# A configuration used for SQLMesh tests with virtual environment mode set to DEV_ONLY.
+test_config_virtual_environment_mode_dev_only = test_config.copy(
+    update={
+        "virtual_environment_mode": VirtualEnvironmentMode.DEV_ONLY,
+        "plan": PlanConfig(
+            auto_categorize_changes=CategorizerConfig.all_full(),
+        ),
+    }
+)
+
 # A DuckDB config with a physical schema map.
 map_config = Config(
     default_connection=DuckDBConnectionConfig(),

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 versioned table names and virtual layer updates.
+    DEV_ONLY: Bypass virtual environments in production, using original unversioned model names.
+    """
+
+    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 (
@@ -110,6 +114,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,12 +153,9 @@ class Config(BaseConfig):
     env_vars: t.Dict[str, str] = {}
     username: str = ""
     physical_schema_mapping: RegexKeyDict = {}
-    environment_suffix_target: EnvironmentSuffixTarget = Field(
-        default=EnvironmentSuffixTarget.default
-    )
-    physical_table_naming_convention: TableNamingConvention = Field(
-        default=TableNamingConvention.default
-    )
+    environment_suffix_target: EnvironmentSuffixTarget = EnvironmentSuffixTarget.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 = {}
@@ -260,6 +262,11 @@ def _normalize_identifiers(key: str) -> None:
                 "Please specify one or the other"
             )
 
+        if self.plan.use_finalized_state and not self.virtual_environment_mode.is_full:
+            raise ConfigError(
+                "Using the finalized state is only supported when `virtual_environment_mode` is set to `full`."
+            )
+
         if self.environment_catalog_mapping:
             _normalize_identifiers("environment_catalog_mapping")
         if self.physical_schema_mapping:

sqlmesh/core/context.py

@@ -1616,6 +1616,11 @@ def plan_builder(
             max_interval_end_per_model,
         )
 
+        if not self.config.virtual_environment_mode.is_full:
+            forward_only = True
+        elif forward_only is None:
+            forward_only = self.config.plan.forward_only
+
         return self.PLAN_BUILDER_TYPE(
             context_diff=context_diff,
             start=start,
@@ -1628,9 +1633,7 @@ def plan_builder(
             skip_backfill=skip_backfill,
             empty_backfill=empty_backfill,
             is_dev=is_dev,
-            forward_only=(
-                forward_only if forward_only is not None else self.config.plan.forward_only
-            ),
+            forward_only=forward_only,
             allow_destructive_models=expanded_destructive_models,
             environment_ttl=environment_ttl,
             environment_suffix_target=self.config.environment_suffix_target,
@@ -2936,7 +2939,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/engine_adapter/base.py

@@ -32,20 +32,21 @@
     CommentCreationTable,
     CommentCreationView,
     DataObject,
+    DataObjectType,
     EngineRunMode,
     InsertOverwriteStrategy,
     SourceQuery,
     set_catalog,
 )
 from sqlmesh.core.model.kind import TimeColumn
 from sqlmesh.core.schema_diff import SchemaDiffer
-from sqlmesh.utils import columns_to_types_all_known, random_id, CorrelationId
-from sqlmesh.utils.connection_pool import create_connection_pool, ConnectionPool
+from sqlmesh.utils import CorrelationId, columns_to_types_all_known, random_id
+from sqlmesh.utils.connection_pool import ConnectionPool, create_connection_pool
 from sqlmesh.utils.date import TimeLike, make_inclusive, to_time_column
 from sqlmesh.utils.errors import (
+    MissingDefaultCatalogError,
     SQLMeshError,
     UnsupportedCatalogOperationError,
-    MissingDefaultCatalogError,
 )
 from sqlmesh.utils.pandas import columns_to_types_from_df
 
@@ -54,8 +55,8 @@
 
     from sqlmesh.core._typing import SchemaName, SessionProperties, TableName
     from sqlmesh.core.engine_adapter._typing import (
-        BigframeSession,
         DF,
+        BigframeSession,
         PySparkDataFrame,
         PySparkSession,
         Query,
@@ -369,6 +370,12 @@ def replace_query(
             kwargs: Optional create table properties.
         """
         target_table = exp.to_table(table_name)
+
+        target_data_object = self.get_data_object(target_table)
+        table_exists = target_data_object is not None
+        if self.drop_data_object_on_type_mismatch(target_data_object, DataObjectType.TABLE):
+            table_exists = False
+
         source_queries, columns_to_types = self._get_source_queries_and_columns_to_types(
             query_or_df, columns_to_types, target_table=target_table
         )
@@ -390,7 +397,7 @@ def replace_query(
             )
         # All engines support `CREATE TABLE AS` so we use that if the table doesn't already exist and we
         # use `CREATE OR REPLACE TABLE AS` if the engine supports it
-        if self.SUPPORTS_REPLACE_TABLE or not self.table_exists(target_table):
+        if self.SUPPORTS_REPLACE_TABLE or not table_exists:
             return self._create_table_from_source_queries(
                 target_table,
                 source_queries,
@@ -930,6 +937,28 @@ def clone_table(
             )
         )
 
+    def drop_data_object(self, data_object: DataObject, ignore_if_not_exists: bool = True) -> None:
+        """Drops a data object of arbitrary type.
+
+        Args:
+            data_object: The data object to drop.
+            ignore_if_not_exists: If True, no error will be raised if the data object does not exist.
+        """
+        if data_object.type.is_view:
+            self.drop_view(data_object.to_table(), ignore_if_not_exists=ignore_if_not_exists)
+        elif data_object.type.is_materialized_view:
+            self.drop_view(
+                data_object.to_table(), ignore_if_not_exists=ignore_if_not_exists, materialized=True
+            )
+        elif data_object.type.is_table:
+            self.drop_table(data_object.to_table(), exists=ignore_if_not_exists)
+        elif data_object.type.is_managed_table:
+            self.drop_managed_table(data_object.to_table(), exists=ignore_if_not_exists)
+        else:
+            raise SQLMeshError(
+                f"Can't drop data object '{data_object.to_table().sql(dialect=self.dialect)}' of type '{data_object.type.value}'"
+            )
+
     def drop_table(self, table_name: TableName, exists: bool = True) -> None:
         """Drops a table.
 
@@ -1118,6 +1147,12 @@ def create_view(
         if properties.expressions:
             create_kwargs["properties"] = properties
 
+        if replace:
+            self.drop_data_object_on_type_mismatch(
+                self.get_data_object(view_name),
+                DataObjectType.VIEW if not materialized else DataObjectType.MATERIALIZED_VIEW,
+            )
+
         with source_queries[0] as query:
             self.execute(
                 exp.Create(
@@ -2022,6 +2057,15 @@ def rename_table(
                 )
         self._rename_table(old_table_name, new_table_name)
 
+    def get_data_object(self, target_name: TableName) -> t.Optional[DataObject]:
+        target_table = exp.to_table(target_name)
+        existing_data_objects = self.get_data_objects(
+            schema_(target_table.db, target_table.catalog), {target_table.name}
+        )
+        if existing_data_objects:
+            return existing_data_objects[0]
+        return None
+
     def get_data_objects(
         self, schema_name: SchemaName, object_names: t.Optional[t.Set[str]] = None
     ) -> t.List[DataObject]:
@@ -2483,6 +2527,30 @@ def _truncate_table(self, table_name: TableName) -> None:
         table = exp.to_table(table_name)
         self.execute(f"TRUNCATE TABLE {table.sql(dialect=self.dialect, identify=True)}")
 
+    def drop_data_object_on_type_mismatch(
+        self, data_object: t.Optional[DataObject], expected_type: DataObjectType
+    ) -> bool:
+        """Drops a data object if it exists and is not of the expected type.
+
+        Args:
+            data_object: The data object to check.
+            expected_type: The expected type of the data object.
+
+        Returns:
+            True if the data object was dropped, False otherwise.
+        """
+        if data_object is None or data_object.type == expected_type:
+            return False
+
+        logger.warning(
+            "Target data object '%s' is a %s and not a %s, dropping it",
+            data_object.to_table().sql(dialect=self.dialect),
+            data_object.type.value,
+            expected_type.value,
+        )
+        self.drop_data_object(data_object)
+        return True
+
     def _replace_by_key(
         self,
         target_table: TableName,

sqlmesh/core/engine_adapter/redshift.py

@@ -262,7 +262,12 @@ def replace_query(
         """
         import pandas as pd
 
-        if not isinstance(query_or_df, pd.DataFrame) or not self.table_exists(table_name):
+        target_data_object = self.get_data_object(table_name)
+        table_exists = target_data_object is not None
+        if self.drop_data_object_on_type_mismatch(target_data_object, DataObjectType.TABLE):
+            table_exists = False
+
+        if not isinstance(query_or_df, pd.DataFrame) or not table_exists:
             return super().replace_query(
                 table_name,
                 query_or_df,

sqlmesh/core/engine_adapter/shared.py

@@ -171,6 +171,9 @@ class DataObject(PydanticModel):
     def is_clustered(self) -> bool:
         return bool(self.clustering_key)
 
+    def to_table(self) -> exp.Table:
+        return exp.table_(self.name, db=self.schema_name, catalog=self.catalog, quoted=True)
+
 
 class CatalogSupport(Enum):
     # The engine has no concept of catalogs

sqlmesh/core/loader.py

@@ -603,6 +603,7 @@ def _load_sql_models(
                 infer_names=self.config.model_naming.infer_names,
                 signal_definitions=signals,
                 default_catalog_per_gateway=self.context.default_catalog_per_gateway,
+                virtual_environment_mode=self.config.virtual_environment_mode,
                 **loading_default_kwargs or {},
             )
 
@@ -683,6 +684,7 @@ def _load_python_models(
                             audit_definitions=audits,
                             signal_definitions=signals,
                             default_catalog_per_gateway=self.context.default_catalog_per_gateway,
+                            virtual_environment_mode=self.config.virtual_environment_mode,
                         ):
                             if model.enabled:
                                 models[model.fqn] = model