fix: Ensure that representative but not deployable snapshots use the Prod physical models consistently

sqlmesh/core/plan/stages.py

@@ -313,7 +313,7 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
         if plan.new_snapshots:
             stages.append(CreateSnapshotRecordsStage(snapshots=plan.new_snapshots))
 
-        snapshots_to_create = self._get_snapshots_to_create(plan, snapshots)
+        snapshots_to_create = self._get_snapshots_to_create(plan, snapshots, deployability_index)
         if snapshots_to_create:
             stages.append(
                 PhysicalLayerSchemaCreationStage(
@@ -629,7 +629,7 @@ def _get_audit_only_snapshots(
         return audit_snapshots
 
     def _get_snapshots_to_create(
-        self, plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot]
+        self, plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot], deployability_index: DeployabilityIndex
     ) -> t.List[Snapshot]:
         promoted_snapshot_ids = (
             set(plan.environment.promoted_snapshot_ids)
@@ -640,6 +640,11 @@ def _get_snapshots_to_create(
         def _should_create(s: Snapshot) -> bool:
             if not s.is_model or s.is_symbolic:
                 return False
+
+            # Do not create snapshots that contain production data but can never be deployed.
+            if deployability_index.is_representative(s) and not deployability_index.is_deployable(s):
+                return False
+
             # Only create tables for snapshots that we're planning to promote or that were selected for backfill
             return (
                 plan.is_selected_for_backfill(s.name)

sqlmesh/core/snapshot/definition.py

@@ -1047,6 +1047,13 @@ def missing_intervals(
             return []
 
         deployability_index = deployability_index or DeployabilityIndex.all_deployable()
+        # If a snapshot is representative but not deployable (e.g. INDIRECT_NON_BREAKING snapshots) then it will only
+        # ever refer to the prod physical table, and so should never have any missing intervals.
+        if deployability_index.is_representative(self) and not deployability_index.is_deployable(
+            self
+        ):
+            return []
+
         intervals = (
             self.intervals if deployability_index.is_representative(self) else self.dev_intervals
         )
@@ -1610,7 +1617,7 @@ def is_deployable(self, snapshot: SnapshotIdLike) -> bool:
 
     def is_representative(self, snapshot: SnapshotIdLike) -> bool:
         """Returns true if the deployable (non-dev) table of the given snapshot should be used for reading, table mapping, and
-        computing missing intervals.
+        computing missing intervals (if the snapshot is also deployable).
 
         Note, that deployable snapshots are also representative, but the reverse is not always true.
 
@@ -2063,6 +2070,13 @@ def missing_intervals(
         if not snapshot.evaluatable:
             continue
 
+        # If a snapshot is representative but not deployable (e.g. INDIRECT_NON_BREAKING snapshots) then it will only
+        # ever refer to the prod physical table, and so should never have any missing intervals.
+        if deployability_index.is_representative(
+            snapshot
+        ) and not deployability_index.is_deployable(snapshot):
+            continue
+
         snapshot_start_date = start_override_per_model.get(snapshot.name, start_dt)
         snapshot_end_date: TimeLike = end_date
 

tests/core/test_snapshot.py

@@ -638,6 +638,71 @@ def test_missing_intervals_start_override_per_model(make_snapshot: t.Callable[..
     ]
 
 
+def test__missing_intervals__representative_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_representative(snapshot)
+    assert missing_intervals == [
+        (to_timestamp("2020-01-03"), to_timestamp("2020-01-04")),
+    ]
+
+
+def test__missing_intervals__non_representative_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_representative(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__missing_intervals__representative_but_non_deployable_snapshot_no_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.INDIRECT_NON_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_representative(snapshot)
+    assert not deployability_index.is_deployable(snapshot)
+    assert missing_intervals == []
+
+
 def test_incremental_time_self_reference(make_snapshot):
     snapshot = make_snapshot(
         SqlModel(
@@ -2407,63 +2472,101 @@ def test_earliest_start_date(sushi_context: Context):
 
 
 def test_deployability_index(make_snapshot):
-    snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("SELECT 1")))
-    snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)
+    # Breaking change - should be both deployable / representative
+    snapshot_breaking = make_snapshot(SqlModel(name="breaking", query=parse_one("SELECT 1")))
+    snapshot_breaking.categorize_as(SnapshotChangeCategory.BREAKING)
 
-    snapshot_b = make_snapshot(SqlModel(name="b", query=parse_one("SELECT 1")))
-    snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
-    snapshot_b.parents = (snapshot_a.snapshot_id,)
+    # Forward only breaking change - cannot be deployable / representative due to forward only
+    snapshot_breaking_forward_only = make_snapshot(
+        SqlModel(name="forward_only", query=parse_one("SELECT 1"))
+    )
+    snapshot_breaking_forward_only.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
+    snapshot_breaking_forward_only.parents = (snapshot_breaking.snapshot_id,)
 
-    snapshot_c = make_snapshot(SqlModel(name="c", query=parse_one("SELECT 1")))
-    snapshot_c.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
-    snapshot_c.parents = (snapshot_b.snapshot_id,)
+    # Indirect breaking - cannot be deployable / representative due to forward only non-representative parent
+    snapshot_indirect_breaking_non_representative_parent = make_snapshot(
+        SqlModel(name="indirect_breaking_non_representative_parent", query=parse_one("SELECT 1"))
+    )
+    snapshot_indirect_breaking_non_representative_parent.categorize_as(
+        SnapshotChangeCategory.INDIRECT_BREAKING
+    )
+    snapshot_indirect_breaking_non_representative_parent.parents = (
+        snapshot_breaking_forward_only.snapshot_id,
+    )
 
-    snapshot_d = make_snapshot(SqlModel(name="d", query=parse_one("SELECT 1")))
-    snapshot_d.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
-    snapshot_d.parents = (snapshot_b.snapshot_id, snapshot_a.snapshot_id)
+    # Indirect breaking - can be deployable / representative due to forward only representative parent
+    snapshot_indirect_breaking_representative_parent = make_snapshot(
+        SqlModel(name="indirect_breaking_representative_parent", query=parse_one("SELECT 1"))
+    )
+    snapshot_indirect_breaking_representative_parent.categorize_as(
+        SnapshotChangeCategory.INDIRECT_BREAKING
+    )
+    snapshot_indirect_breaking_representative_parent.parents = (snapshot_breaking.snapshot_id,)
 
-    snapshot_e = make_snapshot(SqlModel(name="e", query=parse_one("SELECT 1")))
-    snapshot_e.categorize_as(SnapshotChangeCategory.NON_BREAKING)
+    # Non breaking - deployable / representative due to no breaking changes
+    snapshot_non_breaking = make_snapshot(
+        SqlModel(name="non_breaking", query=parse_one("SELECT 1"))
+    )
+    snapshot_non_breaking.categorize_as(SnapshotChangeCategory.NON_BREAKING)
 
-    snapshot_f = make_snapshot(SqlModel(name="f", query=parse_one("SELECT 1")))
-    snapshot_f.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
-    snapshot_f.parents = (snapshot_e.snapshot_id, snapshot_a.snapshot_id)
+    # Indirect non breaking - can be representative but not deployable
+    snapshot_indirect_non_breaking = make_snapshot(
+        SqlModel(name="indirect_non_breaking", query=parse_one("SELECT 1"))
+    )
+    snapshot_indirect_non_breaking.intervals = [
+        (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
+    ]
+    snapshot_indirect_non_breaking.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
+    snapshot_indirect_non_breaking.parents = (snapshot_non_breaking.snapshot_id,)
 
-    snapshot_g = make_snapshot(SqlModel(name="g", query=parse_one("SELECT 1")))
-    snapshot_g.intervals = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
-    snapshot_g.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
-    snapshot_g.parents = (snapshot_e.snapshot_id,)
+    # Breaking with non-representative parent - cannot be deployable due to non-representative parent
+    snapshot_breaking_non_representative_parent = make_snapshot(
+        SqlModel(name="breaking_non_deployable_parents", query=parse_one("SELECT 1"))
+    )
+    snapshot_breaking_non_representative_parent.categorize_as(SnapshotChangeCategory.BREAKING)
+    snapshot_breaking_non_representative_parent.parents = (
+        snapshot_breaking_forward_only.snapshot_id,
+    )
 
     snapshots = {
         s.snapshot_id: s
         for s in [
-            snapshot_a,
-            snapshot_b,
-            snapshot_c,
-            snapshot_d,
-            snapshot_e,
-            snapshot_f,
-            snapshot_g,
+            snapshot_breaking,
+            snapshot_breaking_forward_only,
+            snapshot_indirect_breaking_non_representative_parent,
+            snapshot_indirect_breaking_representative_parent,
+            snapshot_non_breaking,
+            snapshot_indirect_non_breaking,
+            snapshot_breaking_non_representative_parent,
         ]
     }
 
     deployability_index = DeployabilityIndex.create(snapshots)
 
-    assert deployability_index.is_deployable(snapshot_a)
-    assert deployability_index.is_deployable(snapshot_e)
-    assert deployability_index.is_deployable(snapshot_f)
-    assert not deployability_index.is_deployable(snapshot_g)
-    assert not deployability_index.is_deployable(snapshot_b)
-    assert not deployability_index.is_deployable(snapshot_c)
-    assert not deployability_index.is_deployable(snapshot_d)
+    assert deployability_index.is_deployable(snapshot_breaking)
+    assert deployability_index.is_representative(snapshot_breaking)
 
-    assert deployability_index.is_representative(snapshot_a)
-    assert deployability_index.is_representative(snapshot_e)
-    assert deployability_index.is_representative(snapshot_f)
-    assert deployability_index.is_representative(snapshot_g)
-    assert not deployability_index.is_representative(snapshot_b)
-    assert not deployability_index.is_representative(snapshot_c)
-    assert not deployability_index.is_representative(snapshot_d)
+    assert not deployability_index.is_deployable(snapshot_breaking_forward_only)
+    assert not deployability_index.is_representative(snapshot_breaking_forward_only)
+
+    assert deployability_index.is_deployable(snapshot_non_breaking)
+    assert deployability_index.is_representative(snapshot_non_breaking)
+
+    assert not deployability_index.is_deployable(
+        snapshot_indirect_breaking_non_representative_parent
+    )
+    assert not deployability_index.is_representative(
+        snapshot_indirect_breaking_non_representative_parent
+    )
+
+    assert deployability_index.is_deployable(snapshot_indirect_breaking_representative_parent)
+    assert deployability_index.is_representative(snapshot_indirect_breaking_representative_parent)
+
+    assert not deployability_index.is_deployable(snapshot_indirect_non_breaking)
+    assert deployability_index.is_representative(snapshot_indirect_non_breaking)
+
+    assert not deployability_index.is_deployable(snapshot_breaking_non_representative_parent)
+    assert not deployability_index.is_representative(snapshot_breaking_non_representative_parent)
 
     all_deployable_index = deployability_index.all_deployable()
     assert all(all_deployable_index.is_deployable(s) for s in snapshots.values())

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__representative__prod_physical(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)
 
     model = SqlModel(
@@ -319,9 +319,15 @@ 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,
+    )
 
+    assert deployability_index.is_representative(snapshot)
     adapter_mock.transaction.assert_called()
     adapter_mock.session.assert_called()
     adapter_mock.create_schema.assert_called_once_with(to_schema("test_schema__test_env"))
@@ -336,6 +342,43 @@ def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot):
     )
 
 
+def test_promote__non_representative__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,
+    )
+
+    assert not deployability_index.is_representative(snapshot)
+    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)