Switched missing_interval and physical model logic to use deployability instead of representativeness for determining whether dev or prod instance of each should be used.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>

Author: davem-bis

sqlmesh/core/snapshot/definition.py

@@ -1048,7 +1048,7 @@ def missing_intervals(
 
         deployability_index = deployability_index or DeployabilityIndex.all_deployable()
         intervals = (
-            self.intervals if deployability_index.is_representative(self) else self.dev_intervals
+            self.intervals if deployability_index.is_deployable(self) else self.dev_intervals
         )
 
         if not self.evaluatable or (self.is_seed and intervals):

sqlmesh/core/snapshot/evaluator.py

@@ -1270,7 +1270,7 @@ def _promote_snapshot(
             if environment_naming_info.gateway_managed
             else self.adapter
         )
-        table_name = snapshot.table_name(deployability_index.is_representative(snapshot))
+        table_name = snapshot.table_name(deployability_index.is_deployable(snapshot))
         view_name = snapshot.qualified_view_name.for_environment(
             environment_naming_info, dialect=adapter.dialect
         )

tests/core/test_snapshot.py

@@ -637,6 +637,43 @@ def test_missing_intervals_start_override_per_model(make_snapshot: t.Callable[..
         (to_timestamp("2023-02-07"), to_timestamp("2023-02-08")),
     ]
 
+def test__missing_intervals__deployable_snapshot_prod_intervals_returned(snapshot: Snapshot):
+    # Arrange
+    snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=False)
+    snapshot.add_interval(start="2020-01-02", end="2020-01-02", is_dev=False)
+    snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=True)
+    
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    deployability_index = DeployabilityIndex.create([snapshot])
+
+    # Act
+    missing_intervals = snapshot.missing_intervals(start="2020-01-01", end="2020-01-03", deployability_index=deployability_index)
+
+    # Assert
+    assert deployability_index.is_deployable(snapshot)
+    assert missing_intervals == [
+        (to_timestamp("2020-01-03"), to_timestamp("2020-01-04")),
+    ]
+    
+def test__missing_intervals__non_deployable_snapshot_dev_intervals_returned(snapshot: Snapshot):
+    # Arrange
+    snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=False)
+    snapshot.add_interval(start="2020-01-02", end="2020-01-02", is_dev=False)
+    snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=True)
+    
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
+    deployability_index = DeployabilityIndex.create([snapshot])
+
+    # Act
+    missing_intervals = snapshot.missing_intervals(start="2020-01-01", end="2020-01-03", deployability_index=deployability_index)
+
+    # Assert
+    assert not deployability_index.is_deployable(snapshot)
+    assert missing_intervals == [
+        (to_timestamp("2020-01-02"), to_timestamp("2020-01-03")), # Not missing from prod intervals
+        (to_timestamp("2020-01-03"), to_timestamp("2020-01-04")),
+    ]
+
 
 def test_incremental_time_self_reference(make_snapshot):
     snapshot = make_snapshot(
@@ -3962,3 +3999,4 @@ def test_snapshot_id_and_version_optional_kind_name():
     assert snapshot.model_kind_name
     assert snapshot.is_incremental_unmanaged
     assert snapshot.full_history_restatement_only
+

tests/core/test_snapshot_evaluator.py

@@ -307,7 +307,7 @@ def increment_stage_counter(evaluator) -> None:
     )
 
 
-def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot):
+def test_promote__deployable__non_dev_physical(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)
 
     model = SqlModel(
@@ -319,8 +319,13 @@ def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot):
 
     snapshot = make_snapshot(model)
     snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    deployability_index = DeployabilityIndex.create([snapshot])
 
-    evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env"))
+    evaluator.promote(
+        target_snapshots=[snapshot],
+        environment_naming_info=EnvironmentNamingInfo(name="test_env"),
+        deployability_index=deployability_index
+    )
 
     adapter_mock.transaction.assert_called()
     adapter_mock.session.assert_called()
@@ -336,6 +341,39 @@ def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot):
     )
 
 
+def test_promote__non_deployable__dev_physical(mocker: MockerFixture, adapter_mock, make_snapshot):
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="a"),
+        storage_format="parquet",
+        query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
+    deployability_index = DeployabilityIndex.create([snapshot])
+
+    evaluator.promote(
+        target_snapshots=[snapshot],
+        environment_naming_info=EnvironmentNamingInfo(name="test_env"),
+        deployability_index=deployability_index
+    )
+
+    adapter_mock.transaction.assert_called()
+    adapter_mock.session.assert_called()
+    adapter_mock.create_schema.assert_called_once_with(to_schema("test_schema__test_env"))
+    adapter_mock.create_view.assert_called_once_with(
+        "test_schema__test_env.test_model",
+        parse_one(
+            f"SELECT * FROM sqlmesh__test_schema.test_schema__test_model__{snapshot.version}__dev"
+        ),
+        table_description=None,
+        column_descriptions=None,
+        view_properties={},
+    )
+
 def test_demote(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)