fix: ensure dev_only VDE always applies grants in production

Author: newtonapple

sqlmesh/core/snapshot/evaluator.py

@@ -922,6 +922,7 @@ def _render_and_insert_snapshot(
         model = snapshot.model
         adapter = self.get_adapter(model.gateway)
         evaluation_strategy = _evaluation_strategy(snapshot, adapter)
+        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
 
         queries_or_dfs = self._render_snapshot_for_evaluation(
             snapshot,
@@ -945,6 +946,7 @@ def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
                     execution_time=execution_time,
                     physical_properties=rendered_physical_properties,
                     render_kwargs=create_render_kwargs,
+                    is_snapshot_deployable=is_snapshot_deployable,
                 )
             else:
                 logger.info(
@@ -967,6 +969,7 @@ def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
                     execution_time=execution_time,
                     physical_properties=rendered_physical_properties,
                     render_kwargs=create_render_kwargs,
+                    is_snapshot_deployable=is_snapshot_deployable,
                 )
 
         # DataFrames, unlike SQL expressions, can provide partial results by yielding dataframes. As a result,
@@ -1175,6 +1178,7 @@ def _migrate_target_table(
                 allow_additive_snapshots=allow_additive_snapshots,
                 ignore_destructive=snapshot.model.on_destructive_change.is_ignore,
                 ignore_additive=snapshot.model.on_additive_change.is_ignore,
+                deployability_index=deployability_index,
             )
         finally:
             if snapshot.is_materialized:
@@ -1224,6 +1228,7 @@ def _promote_snapshot(
                 model=snapshot.model,
                 environment=environment_naming_info.name,
                 snapshots=snapshots,
+                snapshot=snapshot,
                 **render_kwargs,
             )
 
@@ -1448,6 +1453,8 @@ def _execute_create(
             is_snapshot_representative=is_snapshot_representative,
             dry_run=dry_run,
             physical_properties=rendered_physical_properties,
+            snapshot=snapshot,
+            deployability_index=deployability_index,
         )
         if run_pre_post_statements:
             adapter.execute(snapshot.model.render_post_statements(**create_render_kwargs))
@@ -1690,6 +1697,7 @@ def _apply_grants(
         model: Model,
         table_name: str,
         target_layer: GrantsTargetLayer,
+        is_snapshot_deployable: bool = False,
     ) -> None:
         """Apply grants for a model if grants are configured.
 
@@ -1701,6 +1709,7 @@ def _apply_grants(
             model: The SQLMesh model containing grants configuration
             table_name: The target table/view name to apply grants to
             target_layer: The grants application layer (physical or virtual)
+            is_snapshot_deployable: Whether the snapshot is deployable (targeting production)
         """
         grants_config = model.grants
         if grants_config is None:
@@ -1714,7 +1723,16 @@ def _apply_grants(
             return
 
         model_grants_target_layer = model.grants_target_layer
-        if not (model_grants_target_layer.is_all or model_grants_target_layer == target_layer):
+
+        is_prod_and_dev_only = is_snapshot_deployable and model.virtual_environment_mode.is_dev_only
+
+        if not (
+            model_grants_target_layer.is_all
+            or model_grants_target_layer == target_layer
+            # Always apply grants in production when VDE is dev_only regardless of target_layer
+            # since only physical tables are created in production
+            or is_prod_and_dev_only
+        ):
             logger.debug(
                 f"Skipping grants application for model {model.name} in {target_layer} layer"
             )
@@ -1832,11 +1850,15 @@ def promote(
             view_properties=model.render_virtual_properties(**render_kwargs),
         )
 
+        snapshot = kwargs["snapshot"]
+        deployability_index = kwargs["deployability_index"]
+        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
+
         # Apply grants to the physical layer (referenced table / view) after promotion
-        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
 
         # Apply grants to the virtual layer (view) after promotion
-        self._apply_grants(model, view_name, GrantsTargetLayer.VIRTUAL)
+        self._apply_grants(model, view_name, GrantsTargetLayer.VIRTUAL, is_snapshot_deployable)
 
     def demote(self, view_name: str, **kwargs: t.Any) -> None:
         logger.info("Dropping view '%s'", view_name)
@@ -1897,7 +1919,10 @@ def create(
 
         # Apply grants after table creation (unless explicitly skipped by caller)
         if not skip_grants:
-            self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+            is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+            self._apply_grants(
+                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+            )
 
     def migrate(
         self,
@@ -1925,7 +1950,13 @@ def migrate(
         self.adapter.alter_table(alter_operations)
 
         # Apply grants after schema migration
-        self._apply_grants(snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL)
+        deployability_index = kwargs.get("deployability_index")
+        is_snapshot_deployable = (
+            deployability_index.is_deployable(snapshot) if deployability_index else False
+        )
+        self._apply_grants(
+            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+        )
 
     def delete(self, name: str, **kwargs: t.Any) -> None:
         _check_table_db_is_physical_schema(name, kwargs["physical_schema"])
@@ -1977,7 +2008,8 @@ def _replace_query_for_model(
 
         # Apply grants after table replacement (unless explicitly skipped by caller)
         if not skip_grants:
-            self._apply_grants(model, name, GrantsTargetLayer.PHYSICAL)
+            is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+            self._apply_grants(model, name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
 
     def _get_target_and_source_columns(
         self,
@@ -2267,7 +2299,10 @@ def create(
 
             if not skip_grants:
                 # Apply grants after seed table creation and data insertion
-                self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+                is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+                self._apply_grants(
+                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+                )
         except Exception:
             self.adapter.drop_table(table_name)
             raise
@@ -2339,7 +2374,10 @@ def create(
 
         if not skip_grants:
             # Apply grants after SCD Type 2 table creation
-            self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+            is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+            self._apply_grants(
+                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+            )
 
     def insert(
         self,
@@ -2465,7 +2503,8 @@ def insert(
         )
 
         # Apply grants after view creation / replacement
-        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+        is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+        self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable)
 
     def append(
         self,
@@ -2486,14 +2525,18 @@ def create(
         skip_grants: bool,
         **kwargs: t.Any,
     ) -> None:
+        is_snapshot_deployable: bool = kwargs["is_snapshot_deployable"]
+
         if self.adapter.table_exists(table_name):
             # Make sure we don't recreate the view to prevent deletion of downstream views in engines with no late
             # binding support (because of DROP CASCADE).
             logger.info("View '%s' already exists", table_name)
 
             if not skip_grants:
                 # Always apply grants when present, even if view exists, to handle grants updates
-                self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+                self._apply_grants(
+                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+                )
             return
 
         logger.info("Creating view '%s'", table_name)
@@ -2519,7 +2562,9 @@ def create(
 
         if not skip_grants:
             # Apply grants after view creation
-            self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+            self._apply_grants(
+                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+            )
 
     def migrate(
         self,
@@ -2548,7 +2593,13 @@ def migrate(
         )
 
         # Apply grants after view migration
-        self._apply_grants(snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL)
+        deployability_index = kwargs.get("deployability_index")
+        is_snapshot_deployable = (
+            deployability_index.is_deployable(snapshot) if deployability_index else False
+        )
+        self._apply_grants(
+            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+        )
 
     def delete(self, name: str, **kwargs: t.Any) -> None:
         cascade = kwargs.pop("cascade", False)
@@ -2718,7 +2769,9 @@ def create(
 
             # Apply grants after managed table creation
             if not skip_grants:
-                self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+                self._apply_grants(
+                    model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+                )
 
         elif not is_table_deployable:
             # Only create the dev preview table as a normal table.
@@ -2758,7 +2811,9 @@ def insert(
                 column_descriptions=model.column_descriptions,
                 table_format=model.table_format,
             )
-            self._apply_grants(model, table_name, GrantsTargetLayer.PHYSICAL)
+            self._apply_grants(
+                model, table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+            )
         elif not is_snapshot_deployable:
             # Snapshot isnt deployable; update the preview table instead
             # If the snapshot was deployable, then data would have already been loaded in create() because a managed table would have been created
@@ -2808,8 +2863,13 @@ def migrate(
             )
 
         # Apply grants after verifying no schema changes
-        # This ensures metadata-only grants changes are applied
-        self._apply_grants(snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL)
+        deployability_index = kwargs.get("deployability_index")
+        is_snapshot_deployable = (
+            deployability_index.is_deployable(snapshot) if deployability_index else False
+        )
+        self._apply_grants(
+            snapshot.model, target_table_name, GrantsTargetLayer.PHYSICAL, is_snapshot_deployable
+        )
 
     def delete(self, name: str, **kwargs: t.Any) -> None:
         # a dev preview table is created as a normal table, so it needs to be dropped as a normal table