Fix: Dont reuse the dev table version for non-materialized models (#3694)

Author: izeigerman

sqlmesh/core/snapshot/definition.py

@@ -926,7 +926,7 @@ def categorize_as(self, category: SnapshotChangeCategory) -> None:
             previous_version = self.previous_version
             self.version = previous_version.data_version.version
             self.physical_schema_ = previous_version.physical_schema
-            if category.is_indirect_non_breaking or category.is_metadata:
+            if self.is_materialized and (category.is_indirect_non_breaking or category.is_metadata):
                 # Reuse the dev table for indirect non-breaking changes.
                 self.temp_version = (
                     previous_version.data_version.temp_version

tests/core/test_snapshot.py

@@ -952,6 +952,8 @@ def test_table_name(snapshot: Snapshot, make_snapshot: t.Callable):
     assert snapshot.table_name(is_deployable=True) == "sqlmesh__default.name__3078928823"
     assert snapshot.table_name(is_deployable=False) == "sqlmesh__default.name__3078928823__temp"
 
+    assert not snapshot.temp_version
+
     # Mimic an indirect non-breaking change.
     previous_data_version = snapshot.data_version
     assert previous_data_version.physical_schema == "sqlmesh__default"
@@ -963,6 +965,7 @@ def test_table_name(snapshot: Snapshot, make_snapshot: t.Callable):
     assert snapshot.table_name(is_deployable=True) == "sqlmesh__default.name__3078928823"
     # Indirect non-breaking snapshots reuse the dev table as well.
     assert snapshot.table_name(is_deployable=False) == "sqlmesh__default.name__3078928823__temp"
+    assert snapshot.temp_version
 
     # Mimic a direct forward-only change.
     snapshot.fingerprint = SnapshotFingerprint(
@@ -993,6 +996,37 @@ def test_table_name(snapshot: Snapshot, make_snapshot: t.Callable):
     )
 
 
+def test_table_name_view(make_snapshot: t.Callable):
+    # Mimic a direct breaking change.
+    snapshot = make_snapshot(SqlModel(name="name", query=parse_one("select 1"), kind="VIEW"))
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    snapshot.previous_versions = ()
+    assert snapshot.table_name(is_deployable=True) == f"sqlmesh__default.name__{snapshot.version}"
+    assert (
+        snapshot.table_name(is_deployable=False)
+        == f"sqlmesh__default.name__{snapshot.temp_version_get_or_generate()}__temp"
+    )
+
+    assert not snapshot.temp_version
+
+    # Mimic an indirect non-breaking change.
+    new_snapshot = make_snapshot(SqlModel(name="name", query=parse_one("select 2"), kind="VIEW"))
+    previous_data_version = snapshot.data_version
+    new_snapshot.previous_versions = (previous_data_version,)
+    new_snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
+    assert (
+        new_snapshot.table_name(is_deployable=True) == f"sqlmesh__default.name__{snapshot.version}"
+    )
+    # Indirect non-breaking view snapshots should not reuse the dev table.
+    assert (
+        new_snapshot.table_name(is_deployable=False)
+        == f"sqlmesh__default.name__{new_snapshot.temp_version_get_or_generate()}__temp"
+    )
+    assert not new_snapshot.temp_version
+    assert new_snapshot.version == snapshot.version
+    assert new_snapshot.temp_version_get_or_generate() != snapshot.temp_version_get_or_generate()
+
+
 def test_categorize_change_sql(make_snapshot):
     old_snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))