more tests

Author: izeigerman

sqlmesh/core/scheduler.py

@@ -480,7 +480,7 @@ def run_merged_intervals(
                 execution_time=execution_time,
             )
 
-        def evaluate_node(node: SchedulingUnit) -> None:
+        def run_node(node: SchedulingUnit) -> None:
             if circuit_breaker and circuit_breaker():
                 raise CircuitBreakerError()
             if isinstance(node, DummyNode):
@@ -546,7 +546,7 @@ def evaluate_node(node: SchedulingUnit) -> None:
             with self.snapshot_evaluator.concurrent_context():
                 errors, skipped_intervals = concurrent_apply_to_dag(
                     dag,
-                    evaluate_node,
+                    run_node,
                     self.max_workers,
                     raise_on_error=False,
                 )

sqlmesh/core/snapshot/evaluator.py

@@ -733,15 +733,7 @@ def _evaluate_snapshot(
             adapter.execute(model.render_pre_statements(**render_statements_kwargs))
 
             if not target_table_exists or (model.is_seed and not snapshot.intervals):
-                if (
-                    not is_snapshot_deployable
-                    and snapshot.is_forward_only
-                    and snapshot.is_materialized
-                    and snapshot.previous_versions
-                    and adapter.SUPPORTS_CLONING
-                    # managed models cannot have their schema mutated because theyre based on queries, so clone + alter wont work
-                    and not snapshot.is_managed
-                ):
+                if self._can_clone(snapshot, deployability_index):
                     self._clone_snapshot_in_dev(
                         snapshot=snapshot,
                         snapshots=snapshots,
@@ -750,18 +742,17 @@ def _evaluate_snapshot(
                         rendered_physical_properties=rendered_physical_properties,
                         allow_destructive_snapshots=allow_destructive_snapshots,
                     )
-                else:
-                    if model.annotated or model.is_seed or model.kind.is_scd_type_2:
-                        self._execute_create(
-                            snapshot=snapshot,
-                            table_name=target_table_name,
-                            is_table_deployable=is_snapshot_deployable,
-                            deployability_index=deployability_index,
-                            create_render_kwargs=create_render_kwargs,
-                            rendered_physical_properties=rendered_physical_properties,
-                            dry_run=False,
-                            run_pre_post_statements=False,
-                        )
+                elif model.annotated or model.is_seed or model.kind.is_scd_type_2:
+                    self._execute_create(
+                        snapshot=snapshot,
+                        table_name=target_table_name,
+                        is_table_deployable=is_snapshot_deployable,
+                        deployability_index=deployability_index,
+                        create_render_kwargs=create_render_kwargs,
+                        rendered_physical_properties=rendered_physical_properties,
+                        dry_run=False,
+                        run_pre_post_statements=False,
+                    )
 
             wap_id: t.Optional[str] = None
             if snapshot.is_materialized and (
@@ -808,6 +799,8 @@ def create_snapshot(
         if not snapshot.is_model:
             return
 
+        logger.info("Creating a physical table for snapshot %s", snapshot.snapshot_id)
+
         adapter = self.get_adapter(snapshot.model.gateway)
         create_render_kwargs: t.Dict[str, t.Any] = dict(
             engine_adapter=adapter,
@@ -824,16 +817,7 @@ def create_snapshot(
                 **create_render_kwargs
             )
 
-            if (
-                snapshot.is_forward_only
-                and snapshot.is_materialized
-                and snapshot.previous_versions
-                and adapter.SUPPORTS_CLONING
-                # managed models cannot have their schema mutated because theyre based on queries, so clone + alter wont work
-                and not snapshot.is_managed
-                # If the deployable table is missing we can't clone it
-                and not deployability_index.is_deployable(snapshot)
-            ):
+            if self._can_clone(snapshot, deployability_index):
                 self._clone_snapshot_in_dev(
                     snapshot=snapshot,
                     snapshots=snapshots,
@@ -1370,6 +1354,19 @@ def _execute_create(
         if run_pre_post_statements:
             adapter.execute(snapshot.model.render_post_statements(**create_render_kwargs))
 
+    def _can_clone(self, snapshot: Snapshot, deployability_index: DeployabilityIndex) -> bool:
+        adapter = self.get_adapter(snapshot.model.gateway)
+        return (
+            snapshot.is_forward_only
+            and snapshot.is_materialized
+            and bool(snapshot.previous_versions)
+            and adapter.SUPPORTS_CLONING
+            # managed models cannot have their schema mutated because theyre based on queries, so clone + alter wont work
+            and not snapshot.is_managed
+            # If the deployable table is missing we can't clone it
+            and not deployability_index.is_deployable(snapshot)
+        )
+
 
 def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) -> EvaluationStrategy:
     klass: t.Type

tests/core/test_integration.py

@@ -37,7 +37,7 @@
     DuckDBConnectionConfig,
     TableNamingConvention,
 )
-from sqlmesh.core.config.common import EnvironmentSuffixTarget
+from sqlmesh.core.config.common import EnvironmentSuffixTarget, VirtualEnvironmentMode
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.context import Context
 from sqlmesh.core.config.categorizer import CategorizerConfig
@@ -2664,6 +2664,66 @@ def test_virtual_environment_mode_dev_only_model_kind_change(init_and_plan_conte
     assert data_objects[0].type == "table"
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_virtual_environment_mode_dev_only_model_kind_change_incremental(
+    init_and_plan_context: t.Callable,
+):
+    context, _ = init_and_plan_context(
+        "examples/sushi", config="test_config_virtual_environment_mode_dev_only"
+    )
+
+    forward_only_model_name = "memory.sushi.test_forward_only_model"
+    forward_only_model_expressions = d.parse(
+        f"""
+        MODEL (
+            name {forward_only_model_name},
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column ds,
+                forward_only true,
+            ),
+        );
+
+        SELECT '2023-01-01' AS ds, 'value' AS value;
+        """
+    )
+    forward_only_model = load_sql_based_model(forward_only_model_expressions)
+    forward_only_model = forward_only_model.copy(
+        update={"virtual_environment_mode": VirtualEnvironmentMode.DEV_ONLY}
+    )
+    context.upsert_model(forward_only_model)
+
+    context.plan("prod", auto_apply=True, no_prompts=True)
+
+    # Change to view
+    model = context.get_model(forward_only_model_name)
+    original_kind = model.kind
+    model = model.copy(update={"kind": ViewKind()})
+    context.upsert_model(model)
+    prod_plan = context.plan_builder("prod", skip_tests=True).build()
+    assert prod_plan.requires_backfill
+    assert prod_plan.missing_intervals
+    assert not prod_plan.context_diff.snapshots[
+        context.get_snapshot(model.name).snapshot_id
+    ].intervals
+    context.apply(prod_plan)
+    data_objects = context.engine_adapter.get_data_objects("sushi", {"test_forward_only_model"})
+    assert len(data_objects) == 1
+    assert data_objects[0].type == "view"
+
+    model = model.copy(update={"kind": original_kind})
+    context.upsert_model(model)
+    prod_plan = context.plan_builder("prod", skip_tests=True).build()
+    assert prod_plan.requires_backfill
+    assert prod_plan.missing_intervals
+    assert not prod_plan.context_diff.snapshots[
+        context.get_snapshot(model.name).snapshot_id
+    ].intervals
+    context.apply(prod_plan)
+    data_objects = context.engine_adapter.get_data_objects("sushi", {"test_forward_only_model"})
+    assert len(data_objects) == 1
+    assert data_objects[0].type == "table"
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_virtual_environment_mode_dev_only_model_kind_change_with_follow_up_changes_in_dev(
     init_and_plan_context: t.Callable,