Fix: Only set effective_from for evaluatable models

sqlmesh/core/model/kind.py

@@ -797,6 +797,7 @@ def supports_python_models(self) -> bool:
 
 class EmbeddedKind(_ModelKind):
     name: Literal[ModelKindName.EMBEDDED] = ModelKindName.EMBEDDED
+    disable_restatement: t.Literal[True] = True
 
     @property
     def supports_python_models(self) -> bool:
@@ -805,6 +806,7 @@ def supports_python_models(self) -> bool:
 
 class ExternalKind(_ModelKind):
     name: Literal[ModelKindName.EXTERNAL] = ModelKindName.EXTERNAL
+    disable_restatement: t.Literal[True] = True
 
 
 class CustomKind(_ModelKind):

sqlmesh/core/plan/builder.py

@@ -600,7 +600,11 @@ def _apply_effective_from(self) -> None:
                 raise PlanError("Effective date cannot be in the future.")
 
         for snapshot in self._context_diff.new_snapshots.values():
-            if not snapshot.disable_restatement and not snapshot.full_history_restatement_only:
+            if (
+                snapshot.evaluatable
+                and not snapshot.disable_restatement
+                and not snapshot.full_history_restatement_only
+            ):
                 snapshot.effective_from = self._effective_from
 
     def _is_forward_only_change(self, s_id: SnapshotId) -> bool:

sqlmesh/core/state_sync/engine_adapter.py

@@ -430,7 +430,10 @@ def unpause_snapshots(
                 unpaused_snapshots[snapshot.unpaused_ts].append(snapshot.snapshot_id)
             elif not is_target_snapshot:
                 target_snapshot = target_snapshots_by_version[(snapshot.name, snapshot.version)]
-                if target_snapshot.normalized_effective_from_ts:
+                if (
+                    target_snapshot.normalized_effective_from_ts
+                    and not target_snapshot.disable_restatement
+                ):
                     # Making sure that there are no overlapping intervals.
                     effective_from_ts = target_snapshot.normalized_effective_from_ts
                     logger.info(

tests/core/test_plan.py

@@ -1290,6 +1290,60 @@ def test_effective_from(make_snapshot, mocker: MockerFixture):
     assert updated_snapshot.effective_from is None
 
 
+def test_effective_from_non_evaluatble_model(make_snapshot, mocker: MockerFixture):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            kind="EMBEDDED",
+            query=parse_one("select 1, ds FROM b"),
+            start="2023-01-01",
+            dialect="duckdb",
+        )
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            kind="EMBEDDED",
+            query=parse_one("select 2, ds FROM b"),
+            start="2023-01-01",
+            dialect="duckdb",
+        )
+    )
+    updated_snapshot.previous_versions = snapshot.all_versions
+
+    context_diff = ContextDiff(
+        environment="test_environment",
+        is_new_environment=True,
+        is_unfinalized_environment=False,
+        normalize_environment_name=True,
+        create_from="prod",
+        added=set(),
+        removed_snapshots={},
+        modified_snapshots={updated_snapshot.name: (updated_snapshot, snapshot)},
+        snapshots={updated_snapshot.snapshot_id: updated_snapshot},
+        new_snapshots={updated_snapshot.snapshot_id: updated_snapshot},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+    )
+
+    schema_differ = DuckDBEngineAdapter.SCHEMA_DIFFER
+    plan_builder = PlanBuilder(
+        context_diff,
+        schema_differ,
+        forward_only=True,
+        start="2023-01-01",
+        end="2023-03-01",
+        is_dev=True,
+    )
+
+    plan_builder.set_effective_from("2023-02-01")
+    assert plan_builder.build().effective_from == "2023-02-01"
+    assert not updated_snapshot.effective_from
+
+
 def test_new_environment_no_changes(make_snapshot, mocker: MockerFixture):
     snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))
     snapshot.categorize_as(SnapshotChangeCategory.BREAKING)