Limit changes to the evaluate command only

Author: izeigerman

sqlmesh/core/context.py

@@ -901,13 +901,7 @@ def get_snapshot(
         """
         if isinstance(node_or_snapshot, Snapshot):
             return node_or_snapshot
-        if isinstance(node_or_snapshot, str) and not self.standalone_audits.get(node_or_snapshot):
-            node_or_snapshot = normalize_model_name(
-                node_or_snapshot,
-                dialect=self.default_dialect,
-                default_catalog=self.default_catalog,
-            )
-        fqn = node_or_snapshot if isinstance(node_or_snapshot, str) else node_or_snapshot.fqn
+        fqn = self._node_or_snapshot_to_fqn(node_or_snapshot)
         snapshot = self.snapshots.get(fqn)
 
         if raise_if_missing and not snapshot:
@@ -1052,14 +1046,30 @@ def evaluate(
             execution_time: The date/time time reference to use for execution time.
             limit: A limit applied to the model.
         """
-        snapshot = self.get_snapshot(model_or_snapshot, raise_if_missing=True)
+        snapshots = self.snapshots
+        fqn = self._node_or_snapshot_to_fqn(model_or_snapshot)
+        if fqn not in snapshots:
+            raise SQLMeshError(f"Cannot find snapshot for '{fqn}'")
+        snapshot = snapshots[fqn]
+
+        # Expand all uncategorized parents since physical tables don't exist for them yet
+        expand = [
+            parent
+            for parent in self.dag.upstream(snapshot.model.fqn)
+            if (parent_snapshot := snapshots.get(parent))
+            and parent_snapshot.is_model
+            and parent_snapshot.model.is_sql
+            and not parent_snapshot.categorized
+        ]
+
         df = self.snapshot_evaluator.evaluate_and_fetch(
             snapshot,
             start=start,
             end=end,
             execution_time=execution_time,
             snapshots=self.snapshots,
             limit=limit or c.DEFAULT_MAX_LIMIT,
+            expand=expand,
         )
 
         if df is None:
@@ -2448,6 +2458,19 @@ def _nodes_to_snapshots(self, nodes: t.Dict[str, Node]) -> t.Dict[str, Snapshot]
             snapshots[snapshot.name] = snapshot
         return snapshots
 
+    def _node_or_snapshot_to_fqn(self, node_or_snapshot: NodeOrSnapshot) -> str:
+        if isinstance(node_or_snapshot, Snapshot):
+            return node_or_snapshot.name
+        if isinstance(node_or_snapshot, str) and not self.standalone_audits.get(node_or_snapshot):
+            return normalize_model_name(
+                node_or_snapshot,
+                dialect=self.default_dialect,
+                default_catalog=self.default_catalog,
+            )
+        if not isinstance(node_or_snapshot, str):
+            return node_or_snapshot.fqn
+        return node_or_snapshot
+
     @property
     def _plan_preview_enabled(self) -> bool:
         if self.config.plan.enable_preview is not None:

sqlmesh/core/renderer.py

@@ -317,10 +317,7 @@ def _resolve_tables(
                 **table_mapping,
             }
             expand = set(expand) | {
-                name
-                for name, snapshot in snapshots.items()
-                if snapshot.is_embedded
-                or (snapshot.is_model and snapshot.model.is_sql and not snapshot.categorized)
+                name for name, snapshot in snapshots.items() if snapshot.is_embedded
             }
 
             if expand:

tests/core/test_context.py

@@ -1470,11 +1470,6 @@ def grant_usage_role(evaluator, schemas, role):
     context = Context(paths=tmp_path, config=config)
     snapshots = {s.name: s for s in context.snapshots.values()}
 
-    from sqlmesh.core.snapshot.definition import SnapshotChangeCategory
-
-    for s in snapshots.values():
-        s.categorize_as(SnapshotChangeCategory.BREAKING)
-
     environment_statements = context._environment_statements[0]
     before_all = environment_statements.before_all
     after_all = environment_statements.after_all
@@ -1501,7 +1496,6 @@ def grant_usage_role(evaluator, schemas, role):
         snapshots=snapshots,
         environment_naming_info=EnvironmentNamingInfo(name="prod"),
         runtime_stage=RuntimeStage.BEFORE_ALL,
-        engine_adapter=context.engine_adapter,
     )
 
     assert after_all_rendered == [

tests/core/test_integration.py

@@ -4076,7 +4076,7 @@ def test_no_backfill_for_model_downstream_of_metadata_change(init_and_plan_conte
 
 
 @time_machine.travel("2023-01-08 15:00:00 UTC")
-def test_evaluate_of_uncategorized_snapshot(init_and_plan_context: t.Callable):
+def test_evaluate_uncategorized_snapshot(init_and_plan_context: t.Callable):
     context, plan = init_and_plan_context("examples/sushi")
     context.apply(plan)
 
@@ -4598,7 +4598,7 @@ def test_multi(mocker):
     )
     assert (
         context.render("silver.d").sql()
-        == '''SELECT "c"."col_a" AS "col_a", 2 AS "two", 'repo_2' AS "dup" FROM (SELECT DISTINCT "a"."col_a" AS "col_a" FROM (SELECT 1 AS "col_a", 'b' AS "col_b", 1 AS "one", 'repo_1' AS "dup") AS "a") AS "c"'''
+        == '''SELECT "c"."col_a" AS "col_a", 2 AS "two", 'repo_2' AS "dup" FROM "memory"."silver"."c" AS "c"'''
     )
     context._new_state_sync().reset(default_catalog=context.default_catalog)
     plan = context.plan_builder().build()