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,10 @@ 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 +643,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 not deployability_index.is_eligible_for_physical_layer(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/scheduler.py

@@ -185,11 +185,20 @@ def merged_missing_intervals(
             ignore_cron=ignore_cron,
             end_bounded=end_bounded,
         )
+
+        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
+
         # Filtering snapshots after computing missing intervals because we need all snapshots in order
         # to correctly infer start dates.
+        #
+        # missing_intervals will already have removed any snapshots that are representative but not
+        # deployable, an additional check is included here as a safeguard.
         if selected_snapshots is not None:
             snapshots_to_intervals = {
-                s: i for s, i in snapshots_to_intervals.items() if s.name in selected_snapshots
+                s: i
+                for s, i in snapshots_to_intervals.items()
+                if s.name in selected_snapshots
+                and deployability_index.is_eligible_for_physical_layer(s)
             }
         return snapshots_to_intervals
 

sqlmesh/core/snapshot/definition.py

@@ -1048,6 +1048,11 @@ 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 not deployability_index.is_eligible_for_physical_layer(self):
+            return []
+
         intervals = (
             self.intervals if deployability_index.is_representative(self) else self.dev_intervals
         )
@@ -1610,10 +1615,9 @@ 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.
+        """Returns true if the deployable (non-dev) table of the given snapshot should be used for reading and table mapping.
 
-        Note, that deployable snapshots are also representative, but the reverse is not always true.
+        Note, that deployable snapshots are always representative, but the reverse is not always true.
 
         Unlike `is_deployable`, this variant also captures FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots that
         are not deployable by their nature but are currently promoted in production. Therefore, it's safe to consider
@@ -1629,6 +1633,23 @@ def is_representative(self, snapshot: SnapshotIdLike) -> bool:
         representative = snapshot_id in self.representative_shared_version_ids
         return representative or self.is_deployable(snapshot)
 
+    def is_eligible_for_physical_layer(self, snapshot: SnapshotIdLike) -> bool:
+        """Returns true if the snapshot eligible for the physical layer. This will be true unless it is is representative
+        (looks like prod) but not deployable (can never become prod), and is used for computing missing intervals and
+        physical layer creation.
+
+        This will only return false for FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots in non-production environments.
+
+        Args:
+            snapshot: The snapshot to check.
+
+        Returns:
+            True if the snapshot is representative and not deployable, False otherwise.
+        """
+        return not self.is_representative(snapshot=snapshot) or self.is_deployable(
+            snapshot=snapshot
+        )
+
     def with_non_deployable(self, snapshot: SnapshotIdLike) -> DeployabilityIndex:
         """Creates a new index with the given snapshot marked as non-deployable."""
         return self._add_snapshot(snapshot, False)
@@ -1666,6 +1687,11 @@ def create(
         start: t.Optional[TimeLike] = None,  # plan start
         start_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
     ) -> DeployabilityIndex:
+        def _snapshot_is_representative(snapshot: Snapshot) -> bool:
+            return not snapshot.is_paused or (
+                snapshot.is_indirect_non_breaking and bool(snapshot.intervals)
+            )
+
         if not isinstance(snapshots, dict):
             snapshots = {s.snapshot_id: s for s in snapshots}
 
@@ -1721,17 +1747,15 @@ def create(
                     # Similarly, if the model depends on past and the start date is not aligned with the
                     # model's start, we should consider this snapshot non-deployable.
                     this_deployable = False
-                    if not snapshot.is_paused or (
-                        snapshot.is_indirect_non_breaking and snapshot.intervals
-                    ):
+                    if _snapshot_is_representative(snapshot):
                         # This snapshot represents what's currently deployed in prod.
                         representative_shared_version_ids.add(node)
                     else:
                         # If the parent is not representative then its children can't be deployable.
                         children_deployable = False
             else:
                 children_deployable = False
-                if not snapshots[node].is_paused:
+                if _snapshot_is_representative(snapshot):
                     representative_shared_version_ids.add(node)
 
             deployability_mapping[node] = this_deployable
@@ -2067,6 +2091,11 @@ 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 not deployability_index.is_eligible_for_physical_layer(snapshot):
+            continue
+
         snapshot_start_date = start_override_per_model.get(snapshot.name, start_dt)
         snapshot_end_date: TimeLike = end_date
 

tests/core/integration/test_config.py

@@ -280,15 +280,23 @@ def get_prod_dev_views(metadata: DuckDBMetadata) -> t.Tuple[t.Set[exp.Table], t.
         dev_views = {x for x in views if x.catalog == "dev_catalog" if x.db in environments_schemas}
         return prod_views, dev_views
 
-    def get_default_catalog_and_non_tables(
+    def get_default_catalog_and_non_sql_objects(
         metadata: DuckDBMetadata, default_catalog: t.Optional[str]
-    ) -> t.Tuple[t.Set[exp.Table], t.Set[exp.Table]]:
+    ) -> t.Tuple[t.Set[exp.Table], t.Set[exp.Table], t.Set[exp.Table], t.Set[exp.Table]]:
         tables = metadata.qualified_tables
+        views = metadata.qualified_views
         user_default_tables = {
             x for x in tables if x.catalog == default_catalog and x.db != "sqlmesh"
         }
+        user_default_views = {
+            x
+            for x in views
+            if x.catalog == default_catalog
+            and x.db not in (list(environments_schemas) + ["sqlmesh"])
+        }
         non_default_tables = {x for x in tables if x.catalog != default_catalog}
-        return user_default_tables, non_default_tables
+        non_default_views = {x for x in views if x.catalog != default_catalog}
+        return user_default_tables, user_default_views, non_default_tables, non_default_views
 
     context, plan = init_and_plan_context(
         "examples/sushi", config="environment_catalog_mapping_config"
@@ -299,11 +307,16 @@ def get_default_catalog_and_non_tables(
     prod_views, dev_views = get_prod_dev_views(metadata)
     (
         user_default_tables,
+        user_default_views,
         non_default_tables,
-    ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
+        non_default_views,
+    ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog)
     assert len(prod_views) == 16
     assert len(dev_views) == 0
-    assert len(user_default_tables) == 15
+    assert (
+        len(user_default_tables) == 15
+    )  # One of the snapshots (top_waiters) is a view as opposed to a table.
+    assert len(user_default_views) == 1
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
         {
@@ -317,12 +330,16 @@ def get_default_catalog_and_non_tables(
     prod_views, dev_views = get_prod_dev_views(metadata)
     (
         user_default_tables,
+        user_default_views,
         non_default_tables,
-    ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
+        non_default_views,
+    ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog)
     assert len(prod_views) == 16
     assert len(dev_views) == 16
-    assert len(user_default_tables) == 16
+    assert len(user_default_tables) == 15
+    assert len(user_default_views) == 1
     assert len(non_default_tables) == 0
+    assert len(non_default_views) == 32
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
         {
@@ -336,12 +353,16 @@ def get_default_catalog_and_non_tables(
     prod_views, dev_views = get_prod_dev_views(metadata)
     (
         user_default_tables,
+        user_default_views,
         non_default_tables,
-    ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
+        non_default_views,
+    ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog)
     assert len(prod_views) == 16
     assert len(dev_views) == 32
-    assert len(user_default_tables) == 16
+    assert len(user_default_tables) == 15
+    assert len(user_default_views) == 1
     assert len(non_default_tables) == 0
+    assert len(non_default_views) == 48
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
         {
@@ -356,12 +377,16 @@ def get_default_catalog_and_non_tables(
     prod_views, dev_views = get_prod_dev_views(metadata)
     (
         user_default_tables,
+        user_default_views,
         non_default_tables,
-    ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
+        non_default_views,
+    ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog)
     assert len(prod_views) == 16
     assert len(dev_views) == 16
-    assert len(user_default_tables) == 16
+    assert len(user_default_tables) == 15
+    assert len(user_default_views) == 1
     assert len(non_default_tables) == 0
+    assert len(non_default_views) == 32
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
         {

tests/core/integration/test_forward_only.py

@@ -62,14 +62,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request
     plan = plan_builder.set_effective_from("2023-01-05").build()
     # Default start should be set to effective_from
     assert plan.missing_intervals == [
-        SnapshotIntervals(
-            snapshot_id=top_waiters_snapshot.snapshot_id,
-            intervals=[
-                (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
-                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
-                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
-            ],
-        ),
+        # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=snapshot.snapshot_id,
             intervals=[
@@ -83,13 +76,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request
     plan = plan_builder.set_start("2023-01-06").build()
     # Start override should take precedence
     assert plan.missing_intervals == [
-        SnapshotIntervals(
-            snapshot_id=top_waiters_snapshot.snapshot_id,
-            intervals=[
-                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
-                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
-            ],
-        ),
+        # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=snapshot.snapshot_id,
             intervals=[
@@ -103,13 +90,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request
     # Start should remain unchanged
     assert plan.start == "2023-01-06"
     assert plan.missing_intervals == [
-        SnapshotIntervals(
-            snapshot_id=top_waiters_snapshot.snapshot_id,
-            intervals=[
-                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
-                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
-            ],
-        ),
+        # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=snapshot.snapshot_id,
             intervals=[
@@ -133,6 +114,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request
     # Make sure that the previously set effective_from is respected
     assert prod_plan.start == to_timestamp("2023-01-04")
     assert prod_plan.missing_intervals == [
+        # top_waiters_snapshot has missing intervals in the production environment despite it being an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=top_waiters_snapshot.snapshot_id,
             intervals=[
@@ -208,18 +190,7 @@ def test_forward_only_model_regular_plan(init_and_plan_context: t.Callable):
     )
     plan_builder.set_start("2023-01-06")
     assert plan_builder.build().missing_intervals == [
-        SnapshotIntervals(
-            snapshot_id=top_waiters_snapshot.snapshot_id,
-            intervals=[
-                (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
-                (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
-                (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
-                (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
-                (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
-                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
-                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
-            ],
-        ),
+        # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=snapshot.snapshot_id,
             intervals=[
@@ -232,18 +203,7 @@ def test_forward_only_model_regular_plan(init_and_plan_context: t.Callable):
     # Make sure that changed start is reflected in missing intervals
     plan_builder.set_start("2023-01-07")
     assert plan_builder.build().missing_intervals == [
-        SnapshotIntervals(
-            snapshot_id=top_waiters_snapshot.snapshot_id,
-            intervals=[
-                (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
-                (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
-                (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
-                (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
-                (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
-                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
-                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
-            ],
-        ),
+        # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change.
         SnapshotIntervals(
             snapshot_id=snapshot.snapshot_id,
             intervals=[