From a4cf88c500475e2af5e15b9580de58d6c050c214 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:57:36 +0100
Subject: [PATCH 01/13] Ensured that representative but not deployable
 snapshots do not have any intervals and are a pure pass-through of prod.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/snapshot/definition.py   |  14 ++
 tests/core/test_snapshot.py           | 185 ++++++++++++++++++++------
 tests/core/test_snapshot_evaluator.py |  47 ++++++-
 3 files changed, 203 insertions(+), 43 deletions(-)

diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 0c9635a7c2..24ab485ec8 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/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
         )
@@ -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
 
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 64bee7f472..02936a6bad 100644
--- a/tests/core/test_snapshot.py
+++ b/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())
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 3950e0d53f..4c0f48efbf 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/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)
 

From 0c03fc22078d4b99e211bf14e35b900e7f33afb2 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:09:32 +0100
Subject: [PATCH 02/13] Updated is_representative docstring.

---
 sqlmesh/core/snapshot/definition.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 24ab485ec8..0d2cc1eed4 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1617,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.
 

From f83d29e063e3dc1271a46b86f51bf3e0fdb25e62 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Wed, 16 Sep 2026 16:54:10 +0100
Subject: [PATCH 03/13] Ensured that representative but non-deployable
 snapshots are not created.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/plan/stages.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index 729e1705b4..0972187faa 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/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)

From e6836f7ec6bba204d0b294e29a086327d452fe77 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Fri, 25 Sep 2026 17:21:57 +0100
Subject: [PATCH 04/13] Added whether or not the model is representative but
 not deployable to the missing interval check.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/scheduler.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 5eb0ff40ff..dad83dfac8 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -180,11 +180,14 @@ 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.
         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 not (deployability_index.is_representative(s) and not deployability_index.is_deployable(s))
             }
         return snapshots_to_intervals
 

From b1a8ba2f7b860dd297006c038ed321dffed01690 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Fri, 25 Sep 2026 17:23:26 +0100
Subject: [PATCH 05/13] make style formatting changes.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/plan/stages.py | 9 +++++++--
 sqlmesh/core/scheduler.py   | 8 +++++++-
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index 0972187faa..15f3694338 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/sqlmesh/core/plan/stages.py
@@ -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], deployability_index: DeployabilityIndex
+        self,
+        plan: EvaluatablePlan,
+        snapshots: t.Dict[SnapshotId, Snapshot],
+        deployability_index: DeployabilityIndex,
     ) -> t.List[Snapshot]:
         promoted_snapshot_ids = (
             set(plan.environment.promoted_snapshot_ids)
@@ -642,7 +645,9 @@ def _should_create(s: Snapshot) -> bool:
                 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):
+            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
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index dad83dfac8..bd14a859ef 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -187,7 +187,13 @@ def merged_missing_intervals(
         # to correctly infer start dates.
         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 and not (deployability_index.is_representative(s) and not deployability_index.is_deployable(s))
+                s: i
+                for s, i in snapshots_to_intervals.items()
+                if s.name in selected_snapshots
+                and not (
+                    deployability_index.is_representative(s)
+                    and not deployability_index.is_deployable(s)
+                )
             }
         return snapshots_to_intervals
 

From 0fa1cade4d1bd85a65402d5b5af600b222f24bfe Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:47:29 +0000
Subject: [PATCH 06/13] Ensured that representativeness is appropriately
 determined in dev only VDE mode.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/snapshot/definition.py | 4 +++-
 sqlmesh/core/snapshot/evaluator.py  | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index b73f5a1f53..8d0bde783d 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1738,7 +1738,9 @@ def create(
                         children_deployable = False
             else:
                 children_deployable = False
-                if not snapshots[node].is_paused:
+                if not snapshot.is_paused or (
+                    snapshot.is_indirect_non_breaking and snapshot.intervals
+                ):
                     representative_shared_version_ids.add(node)
 
             deployability_mapping[node] = this_deployable
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 345f8816e6..6a53b96d59 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1574,7 +1574,7 @@ def _execute_create(
     def _can_clone(self, snapshot: Snapshot, deployability_index: DeployabilityIndex) -> bool:
         adapter = self.get_adapter(snapshot.model.gateway)
         return (
-            snapshot.is_forward_only
+            deployability_index.is_representative(snapshot)
             and snapshot.is_materialized
             and bool(snapshot.previous_versions)
             and adapter.SUPPORTS_CLONING

From e7b7b995eefd7befaa9746b401380a45794f3ea8 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Tue, 3 Nov 2026 17:28:56 +0000
Subject: [PATCH 07/13] Reverted change to clonability as not being used by
 indirect_non_breaking changes.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/snapshot/evaluator.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 6a53b96d59..345f8816e6 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1574,7 +1574,7 @@ def _execute_create(
     def _can_clone(self, snapshot: Snapshot, deployability_index: DeployabilityIndex) -> bool:
         adapter = self.get_adapter(snapshot.model.gateway)
         return (
-            deployability_index.is_representative(snapshot)
+            snapshot.is_forward_only
             and snapshot.is_materialized
             and bool(snapshot.previous_versions)
             and adapter.SUPPORTS_CLONING

From 1cbdecc6ab3d42468269a91d1525757520226596 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Wed, 4 Nov 2026 14:45:31 +0000
Subject: [PATCH 08/13] Added test to plan stages to ensure that indirect non
 breaking changes do not trigger a backfill.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/snapshot/definition.py |  13 ++--
 tests/core/test_plan_stages.py      | 106 ++++++++++++++++++++++++++++
 tests/utils/test_concurrency.py     |   2 +-
 3 files changed, 114 insertions(+), 7 deletions(-)

diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 8d0bde783d..3d46a61410 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1673,6 +1673,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 snapshot.intervals
+            )
+
         if not isinstance(snapshots, dict):
             snapshots = {s.snapshot_id: s for s in snapshots}
 
@@ -1728,9 +1733,7 @@ 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:
@@ -1738,9 +1741,7 @@ def create(
                         children_deployable = False
             else:
                 children_deployable = False
-                if not snapshot.is_paused or (
-                    snapshot.is_indirect_non_breaking and snapshot.intervals
-                ):
+                if _snapshot_is_representative(snapshot):
                     representative_shared_version_ids.add(node)
 
             deployability_mapping[node] = this_deployable
diff --git a/tests/core/test_plan_stages.py b/tests/core/test_plan_stages.py
index eb3f965761..76e95de549 100644
--- a/tests/core/test_plan_stages.py
+++ b/tests/core/test_plan_stages.py
@@ -78,6 +78,34 @@ def snapshot_c(make_snapshot, snapshot_a: Snapshot) -> Snapshot:
     return snapshot
 
 
+@pytest.fixture
+def snapshot_d(make_snapshot, snapshot_a: Snapshot) -> Snapshot:
+    snapshot = make_snapshot(
+        SqlModel(
+            name="d",
+            query=parse_one("select * from a"),
+            kind=dict(name=ModelKindName.VIEW),
+        ),
+        nodes={'"a"': snapshot_a.model},
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.NON_BREAKING)
+    return snapshot
+
+
+@pytest.fixture
+def snapshot_e(make_snapshot, snapshot_d: Snapshot) -> Snapshot:
+    snapshot = make_snapshot(
+        SqlModel(
+            name="e",
+            query=parse_one("select * from d"),
+            kind=dict(name=ModelKindName.VIEW),
+        ),
+        nodes={'"d"': snapshot_d.model},
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
+    return snapshot
+
+
 def test_build_plan_stages_basic(
     snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture
 ) -> None:
@@ -1688,6 +1716,84 @@ def test_build_plan_stages_indirect_non_breaking_view_migration(
     assert isinstance(stages[8], FinalizeEnvironmentStage)
 
 
+def test_build_plan_stages_indirect_non_breaking_view_dev(
+    snapshot_d: Snapshot, snapshot_e: Snapshot, make_snapshot, mocker: MockerFixture
+) -> None:
+    # Create environment
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {}
+    existing_environment = Environment(
+        name="prod",
+        snapshots=[snapshot_d.table_info, snapshot_e.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="previous_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_d.snapshot_id, snapshot_e.snapshot_id],
+        finalized_ts=to_timestamp("2023-01-02"),
+    )
+    state_reader.get_environment.return_value = existing_environment
+
+    environment = Environment(
+        name="dev",
+        snapshots=[snapshot_d.table_info, snapshot_e.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id="previous_plan",
+        promoted_snapshot_ids=[snapshot_d.snapshot_id, snapshot_e.snapshot_id],
+    )
+
+    # Create evaluatable plan
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[snapshot_d, snapshot_e],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        empty_backfill=False,
+        restatements={},
+        restate_all_snapshots=False,
+        is_dev=False,
+        allow_destructive_models=set(),
+        allow_additive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
+        ignore_cron=False,
+        directly_modified_snapshots=[snapshot_d.snapshot_id],
+        indirectly_modified_snapshots={
+            snapshot_d.name: [snapshot_e.snapshot_id],
+        },
+        metadata_updated_snapshots=[],
+        removed_snapshots=[],
+        requires_backfill=True,
+        models_to_backfill=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    # Build plan stages
+    stages = build_plan_stages(plan, state_reader, None)
+
+    # Verify stages
+    assert len(stages) == 7
+
+    assert isinstance(stages[0], CreateSnapshotRecordsStage)
+    assert isinstance(stages[1], PhysicalLayerSchemaCreationStage)
+    assert isinstance(stages[2], BackfillStage)
+    assert isinstance(stages[3], EnvironmentRecordUpdateStage)
+    assert isinstance(stages[4], BackfillStage)
+    assert isinstance(stages[5], VirtualLayerUpdateStage)
+    assert isinstance(stages[6], FinalizeEnvironmentStage)
+
+    ## Verify that indirect non breaking change does not trigger a backfill
+    assert stages[2].selected_snapshot_ids == {snapshot_d.snapshot_id}
+
+
 def test_build_plan_stages_virtual_environment_mode_filtering(
     make_snapshot, mocker: MockerFixture
 ) -> None:
diff --git a/tests/utils/test_concurrency.py b/tests/utils/test_concurrency.py
index 0d45a375f1..04f9748fd7 100644
--- a/tests/utils/test_concurrency.py
+++ b/tests/utils/test_concurrency.py
@@ -168,7 +168,7 @@ def _run_bounded_python(program: str) -> None:
         env=env,
         capture_output=True,
         text=True,
-        timeout=8,
+        timeout=16,
     )
     assert result.returncode == 0, f"{result.stdout}\n{result.stderr}"
 

From ffb5c62dafc6239c5307293db07a74f82ac26835 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Wed, 4 Nov 2026 15:01:39 +0000
Subject: [PATCH 09/13] Explicit casting of list in return value to satisfy
 MyPy typing requirements.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/snapshot/definition.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 3d46a61410..6efdedd220 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1675,7 +1675,7 @@ def create(
     ) -> DeployabilityIndex:
         def _snapshot_is_representative(snapshot: Snapshot) -> bool:
             return not snapshot.is_paused or (
-                snapshot.is_indirect_non_breaking and snapshot.intervals
+                snapshot.is_indirect_non_breaking and bool(snapshot.intervals)
             )
 
         if not isinstance(snapshots, dict):

From 528cd3b408f696c707b3d337eb6271349dbfe654 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Wed, 4 Nov 2026 16:25:54 +0000
Subject: [PATCH 10/13] Updated integration tests to deal with change in
 missing interval logic.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 tests/core/integration/test_config.py       | 48 ++++++++++++++----
 tests/core/integration/test_forward_only.py | 56 ++++-----------------
 2 files changed, 47 insertions(+), 57 deletions(-)

diff --git a/tests/core/integration/test_config.py b/tests/core/integration/test_config.py
index 5d571cd7c5..a0fc0eeddd 100644
--- a/tests/core/integration/test_config.py
+++ b/tests/core/integration/test_config.py
@@ -280,30 +280,44 @@ 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"
     )
+    logging.getLogger("sqlmesh").setLevel(logging.ERROR)
     context.apply(plan)
     metadata = DuckDBMetadata(context.engine_adapter)
     state_metadata = DuckDBMetadata.from_context(context.state_sync.state_sync)
     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 +331,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 +354,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 +378,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(
         {
diff --git a/tests/core/integration/test_forward_only.py b/tests/core/integration/test_forward_only.py
index 2dddf18efd..96045f7a35 100644
--- a/tests/core/integration/test_forward_only.py
+++ b/tests/core/integration/test_forward_only.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import logging
+
 import typing as t
 import numpy as np  # noqa: TID253
 import pandas as pd  # noqa: TID253
@@ -23,6 +25,8 @@
 
 pytestmark = pytest.mark.slow
 
+logging.getLogger("sqlmesh").setLevel(logging.ERROR)
+
 
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 @pytest.mark.parametrize(
@@ -62,14 +66,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 +80,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 +94,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 +118,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 +194,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 +207,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=[

From 1b702f32753f87b8bca4f143d153b76676b44625 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Wed, 4 Nov 2026 16:45:47 +0000
Subject: [PATCH 11/13] Removed log suppression that was used during testing.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 tests/core/integration/test_forward_only.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/tests/core/integration/test_forward_only.py b/tests/core/integration/test_forward_only.py
index 96045f7a35..a9ba184fd2 100644
--- a/tests/core/integration/test_forward_only.py
+++ b/tests/core/integration/test_forward_only.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import logging
-
 import typing as t
 import numpy as np  # noqa: TID253
 import pandas as pd  # noqa: TID253
@@ -25,8 +23,6 @@
 
 pytestmark = pytest.mark.slow
 
-logging.getLogger("sqlmesh").setLevel(logging.ERROR)
-
 
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 @pytest.mark.parametrize(

From a37847d6441c0af00ac27ad8742eb2a6c27a79c3 Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Thu, 5 Nov 2026 12:46:39 +0000
Subject: [PATCH 12/13] Changes in response to review comments.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/plan/stages.py           |  4 +-
 sqlmesh/core/scheduler.py             |  8 ++--
 sqlmesh/core/snapshot/definition.py   | 29 ++++++++----
 tests/core/integration/test_config.py |  1 -
 tests/core/test_plan_stages.py        | 17 ++++---
 tests/core/test_snapshot.py           | 65 ++++++++++++++++++++-------
 6 files changed, 85 insertions(+), 39 deletions(-)

diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index 15f3694338..d8494dba74 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/sqlmesh/core/plan/stages.py
@@ -645,9 +645,7 @@ def _should_create(s: Snapshot) -> bool:
                 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
-            ):
+            if deployability_index.is_representative_not_deployable(s):
                 return False
 
             # Only create tables for snapshots that we're planning to promote or that were selected for backfill
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index d8c1263f3d..688522c625 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -190,15 +190,15 @@ def merged_missing_intervals(
 
         # 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
-                and not (
-                    deployability_index.is_representative(s)
-                    and not deployability_index.is_deployable(s)
-                )
+                and not deployability_index.is_representative_not_deployable(s)
             }
         return snapshots_to_intervals
 
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 6efdedd220..57a9d9b1e9 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1050,9 +1050,7 @@ def missing_intervals(
         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
-        ):
+        if deployability_index.is_representative_not_deployable(self):
             return []
 
         intervals = (
@@ -1617,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 (if the snapshot is also deployable).
+        """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
@@ -1636,6 +1633,22 @@ 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_representative_not_deployable(self, snapshot: SnapshotIdLike) -> bool:
+        """Returns true if the snapshot is representative (looks like prod) but not deployable (can never become prod), used for
+        computing missing intervals (if the snapshot is also deployable).
+
+        This will only return true 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 self.is_representative(snapshot=snapshot) and not 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)
@@ -2079,9 +2092,7 @@ def missing_intervals(
 
         # 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):
+        if deployability_index.is_representative_not_deployable(snapshot):
             continue
 
         snapshot_start_date = start_override_per_model.get(snapshot.name, start_dt)
diff --git a/tests/core/integration/test_config.py b/tests/core/integration/test_config.py
index a0fc0eeddd..4bc820629d 100644
--- a/tests/core/integration/test_config.py
+++ b/tests/core/integration/test_config.py
@@ -301,7 +301,6 @@ def get_default_catalog_and_non_sql_objects(
     context, plan = init_and_plan_context(
         "examples/sushi", config="environment_catalog_mapping_config"
     )
-    logging.getLogger("sqlmesh").setLevel(logging.ERROR)
     context.apply(plan)
     metadata = DuckDBMetadata(context.engine_adapter)
     state_metadata = DuckDBMetadata.from_context(context.state_sync.state_sync)
diff --git a/tests/core/test_plan_stages.py b/tests/core/test_plan_stages.py
index 76e95de549..a6bb82e237 100644
--- a/tests/core/test_plan_stages.py
+++ b/tests/core/test_plan_stages.py
@@ -1719,6 +1719,9 @@ def test_build_plan_stages_indirect_non_breaking_view_migration(
 def test_build_plan_stages_indirect_non_breaking_view_dev(
     snapshot_d: Snapshot, snapshot_e: Snapshot, make_snapshot, mocker: MockerFixture
 ) -> None:
+    # Setup Snapshots
+    snapshot_e.intervals = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
+
     # Create environment
     state_reader = mocker.Mock(spec=StateReader)
     state_reader.get_snapshots.return_value = {}
@@ -1755,7 +1758,7 @@ def test_build_plan_stages_indirect_non_breaking_view_dev(
         empty_backfill=False,
         restatements={},
         restate_all_snapshots=False,
-        is_dev=False,
+        is_dev=True,
         allow_destructive_models=set(),
         allow_additive_models=set(),
         forward_only=False,
@@ -1780,18 +1783,18 @@ def test_build_plan_stages_indirect_non_breaking_view_dev(
     stages = build_plan_stages(plan, state_reader, None)
 
     # Verify stages
-    assert len(stages) == 7
+    assert len(stages) == 6
 
     assert isinstance(stages[0], CreateSnapshotRecordsStage)
     assert isinstance(stages[1], PhysicalLayerSchemaCreationStage)
     assert isinstance(stages[2], BackfillStage)
     assert isinstance(stages[3], EnvironmentRecordUpdateStage)
-    assert isinstance(stages[4], BackfillStage)
-    assert isinstance(stages[5], VirtualLayerUpdateStage)
-    assert isinstance(stages[6], FinalizeEnvironmentStage)
+    assert isinstance(stages[4], VirtualLayerUpdateStage)
+    assert isinstance(stages[5], FinalizeEnvironmentStage)
 
-    ## Verify that indirect non breaking change does not trigger a backfill
-    assert stages[2].selected_snapshot_ids == {snapshot_d.snapshot_id}
+    ## Verify that indirect non breaking change does not trigger a schema creation or backfill
+    assert snapshot_e not in stages[1].snapshots
+    assert snapshot_e not in stages[2].snapshot_to_intervals
 
 
 def test_build_plan_stages_virtual_environment_mode_filtering(
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 02936a6bad..ad48cf64ac 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -2545,12 +2545,15 @@ def test_deployability_index(make_snapshot):
 
     assert deployability_index.is_deployable(snapshot_breaking)
     assert deployability_index.is_representative(snapshot_breaking)
+    assert not deployability_index.is_representative_not_deployable(snapshot_breaking)
 
     assert not deployability_index.is_deployable(snapshot_breaking_forward_only)
     assert not deployability_index.is_representative(snapshot_breaking_forward_only)
+    assert not deployability_index.is_representative_not_deployable(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_representative_not_deployable(snapshot_non_breaking)
 
     assert not deployability_index.is_deployable(
         snapshot_indirect_breaking_non_representative_parent
@@ -2558,23 +2561,39 @@ def test_deployability_index(make_snapshot):
     assert not deployability_index.is_representative(
         snapshot_indirect_breaking_non_representative_parent
     )
+    assert not deployability_index.is_representative_not_deployable(
+        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_representative_not_deployable(
+        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 deployability_index.is_representative_not_deployable(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)
+    assert not deployability_index.is_representative_not_deployable(
+        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())
     assert all(all_deployable_index.is_representative(s) for s in snapshots.values())
+    assert all(
+        not all_deployable_index.is_representative_not_deployable(s) for s in snapshots.values()
+    )
 
     none_deployable_index = deployability_index.none_deployable()
     assert all(not none_deployable_index.is_deployable(s) for s in snapshots.values())
     assert all(not none_deployable_index.is_representative(s) for s in snapshots.values())
+    assert all(
+        not none_deployable_index.is_representative_not_deployable(s) for s in snapshots.values()
+    )
 
 
 def test_deployability_index_unpaused_forward_only(make_snapshot):
@@ -2586,15 +2605,18 @@ def test_deployability_index_unpaused_forward_only(make_snapshot):
     snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING)
     snapshot_b.parents = (snapshot_a.snapshot_id,)
 
-    deplyability_index = DeployabilityIndex.create(
+    deployability_index = DeployabilityIndex.create(
         {s.snapshot_id: s for s in [snapshot_a, snapshot_b]}
     )
 
-    assert not deplyability_index.is_deployable(snapshot_a)
-    assert deplyability_index.is_deployable(snapshot_b)
+    assert not deployability_index.is_deployable(snapshot_a)
+    assert deployability_index.is_deployable(snapshot_b)
+
+    assert deployability_index.is_representative(snapshot_a)
+    assert deployability_index.is_representative(snapshot_b)
 
-    assert deplyability_index.is_representative(snapshot_a)
-    assert deplyability_index.is_representative(snapshot_b)
+    assert deployability_index.is_representative_not_deployable(snapshot_a)
+    assert not deployability_index.is_representative_not_deployable(snapshot_b)
 
 
 def test_deployability_index_unpaused_auto_restatement(make_snapshot):
@@ -2621,17 +2643,21 @@ def test_deployability_index_unpaused_auto_restatement(make_snapshot):
     snapshot_c.categorize_as(SnapshotChangeCategory.BREAKING)
     snapshot_c.parents = (snapshot_b.snapshot_id,)
 
-    deplyability_index = DeployabilityIndex.create(
+    deployability_index = DeployabilityIndex.create(
         {s.snapshot_id: s for s in [snapshot_a, snapshot_b, snapshot_c]}
     )
 
-    assert not deplyability_index.is_deployable(snapshot_a)
-    assert not deplyability_index.is_deployable(snapshot_b)
-    assert not deplyability_index.is_deployable(snapshot_c)
+    assert not deployability_index.is_deployable(snapshot_a)
+    assert not deployability_index.is_deployable(snapshot_b)
+    assert not deployability_index.is_deployable(snapshot_c)
 
-    assert deplyability_index.is_representative(snapshot_a)
-    assert deplyability_index.is_representative(snapshot_b)
-    assert not deplyability_index.is_representative(snapshot_c)
+    assert deployability_index.is_representative(snapshot_a)
+    assert deployability_index.is_representative(snapshot_b)
+    assert not deployability_index.is_representative(snapshot_c)
+
+    assert deployability_index.is_representative_not_deployable(snapshot_a)
+    assert deployability_index.is_representative_not_deployable(snapshot_b)
+    assert not deployability_index.is_representative_not_deployable(snapshot_c)
 
 
 def test_deployability_index_uncategorized_forward_only_model(make_snapshot):
@@ -2660,6 +2686,9 @@ def test_deployability_index_uncategorized_forward_only_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
+    assert not deployability_index.is_representative_not_deployable(snapshot_a)
+    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+
 
 def test_deployability_index_auto_restatement_model(make_snapshot):
     model_a = SqlModel(
@@ -2689,6 +2718,9 @@ def test_deployability_index_auto_restatement_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
+    assert not deployability_index.is_representative_not_deployable(snapshot_a)
+    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+
 
 def test_deployability_index_categorized_forward_only_model(make_snapshot):
     model_a = SqlModel(
@@ -2718,6 +2750,9 @@ def test_deployability_index_categorized_forward_only_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
+    assert not deployability_index.is_representative_not_deployable(snapshot_a)
+    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+
 
 def test_deployability_index_missing_parent(make_snapshot):
     snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("SELECT 1")))
@@ -2727,10 +2762,10 @@ def test_deployability_index_missing_parent(make_snapshot):
     snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
     snapshot_b.parents = (snapshot_a.snapshot_id,)
 
-    deplyability_index = DeployabilityIndex.create({snapshot_b.snapshot_id: snapshot_b})
+    deployability_index = DeployabilityIndex.create({snapshot_b.snapshot_id: snapshot_b})
 
-    assert not deplyability_index.is_deployable(snapshot_b)
-    assert not deplyability_index.is_deployable(snapshot_a)
+    assert not deployability_index.is_deployable(snapshot_b)
+    assert not deployability_index.is_deployable(snapshot_a)
 
 
 @pytest.mark.parametrize(

From 960603b6fd09589a0663804a0e3240208b9585bc Mon Sep 17 00:00:00 2001
From: davem-bis <68955845+davem-bis@users.noreply.github.com>
Date: Thu, 5 Nov 2026 14:50:44 +0000
Subject: [PATCH 13/13] Renamed method to represent it's purpose as opposed to
 mechanism.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>
---
 sqlmesh/core/plan/stages.py         |  2 +-
 sqlmesh/core/scheduler.py           |  2 +-
 sqlmesh/core/snapshot/definition.py | 15 ++++++-----
 tests/core/test_snapshot.py         | 40 ++++++++++++++---------------
 4 files changed, 30 insertions(+), 29 deletions(-)

diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index d8494dba74..e0e3955d57 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/sqlmesh/core/plan/stages.py
@@ -645,7 +645,7 @@ def _should_create(s: Snapshot) -> bool:
                 return False
 
             # Do not create snapshots that contain production data but can never be deployed.
-            if deployability_index.is_representative_not_deployable(s):
+            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
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 688522c625..6e2903c664 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -198,7 +198,7 @@ def merged_missing_intervals(
                 s: i
                 for s, i in snapshots_to_intervals.items()
                 if s.name in selected_snapshots
-                and not deployability_index.is_representative_not_deployable(s)
+                and deployability_index.is_eligible_for_physical_layer(s)
             }
         return snapshots_to_intervals
 
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 57a9d9b1e9..723884859a 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1050,7 +1050,7 @@ def missing_intervals(
         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_not_deployable(self):
+        if not deployability_index.is_eligible_for_physical_layer(self):
             return []
 
         intervals = (
@@ -1633,11 +1633,12 @@ 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_representative_not_deployable(self, snapshot: SnapshotIdLike) -> bool:
-        """Returns true if the snapshot is representative (looks like prod) but not deployable (can never become prod), used for
-        computing missing intervals (if the snapshot is also deployable).
+    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 true for FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots in non-production environments.
+        This will only return false for FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots in non-production environments.
 
         Args:
             snapshot: The snapshot to check.
@@ -1645,7 +1646,7 @@ def is_representative_not_deployable(self, snapshot: SnapshotIdLike) -> bool:
         Returns:
             True if the snapshot is representative and not deployable, False otherwise.
         """
-        return self.is_representative(snapshot=snapshot) and not self.is_deployable(
+        return not self.is_representative(snapshot=snapshot) or self.is_deployable(
             snapshot=snapshot
         )
 
@@ -2092,7 +2093,7 @@ def missing_intervals(
 
         # 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_not_deployable(snapshot):
+        if not deployability_index.is_eligible_for_physical_layer(snapshot):
             continue
 
         snapshot_start_date = start_override_per_model.get(snapshot.name, start_dt)
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index ad48cf64ac..2ec37c8a5a 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -2545,15 +2545,15 @@ def test_deployability_index(make_snapshot):
 
     assert deployability_index.is_deployable(snapshot_breaking)
     assert deployability_index.is_representative(snapshot_breaking)
-    assert not deployability_index.is_representative_not_deployable(snapshot_breaking)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_breaking)
 
     assert not deployability_index.is_deployable(snapshot_breaking_forward_only)
     assert not deployability_index.is_representative(snapshot_breaking_forward_only)
-    assert not deployability_index.is_representative_not_deployable(snapshot_breaking_forward_only)
+    assert deployability_index.is_eligible_for_physical_layer(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_representative_not_deployable(snapshot_non_breaking)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_non_breaking)
 
     assert not deployability_index.is_deployable(
         snapshot_indirect_breaking_non_representative_parent
@@ -2561,23 +2561,23 @@ def test_deployability_index(make_snapshot):
     assert not deployability_index.is_representative(
         snapshot_indirect_breaking_non_representative_parent
     )
-    assert not deployability_index.is_representative_not_deployable(
+    assert deployability_index.is_eligible_for_physical_layer(
         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_representative_not_deployable(
+    assert deployability_index.is_eligible_for_physical_layer(
         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 deployability_index.is_representative_not_deployable(snapshot_indirect_non_breaking)
+    assert not deployability_index.is_eligible_for_physical_layer(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)
-    assert not deployability_index.is_representative_not_deployable(
+    assert deployability_index.is_eligible_for_physical_layer(
         snapshot_breaking_non_representative_parent
     )
 
@@ -2585,14 +2585,14 @@ def test_deployability_index(make_snapshot):
     assert all(all_deployable_index.is_deployable(s) for s in snapshots.values())
     assert all(all_deployable_index.is_representative(s) for s in snapshots.values())
     assert all(
-        not all_deployable_index.is_representative_not_deployable(s) for s in snapshots.values()
+        all_deployable_index.is_eligible_for_physical_layer(s) for s in snapshots.values()
     )
 
     none_deployable_index = deployability_index.none_deployable()
     assert all(not none_deployable_index.is_deployable(s) for s in snapshots.values())
     assert all(not none_deployable_index.is_representative(s) for s in snapshots.values())
     assert all(
-        not none_deployable_index.is_representative_not_deployable(s) for s in snapshots.values()
+        none_deployable_index.is_eligible_for_physical_layer(s) for s in snapshots.values()
     )
 
 
@@ -2615,8 +2615,8 @@ def test_deployability_index_unpaused_forward_only(make_snapshot):
     assert deployability_index.is_representative(snapshot_a)
     assert deployability_index.is_representative(snapshot_b)
 
-    assert deployability_index.is_representative_not_deployable(snapshot_a)
-    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+    assert not deployability_index.is_eligible_for_physical_layer(snapshot_a)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_b)
 
 
 def test_deployability_index_unpaused_auto_restatement(make_snapshot):
@@ -2655,9 +2655,9 @@ def test_deployability_index_unpaused_auto_restatement(make_snapshot):
     assert deployability_index.is_representative(snapshot_b)
     assert not deployability_index.is_representative(snapshot_c)
 
-    assert deployability_index.is_representative_not_deployable(snapshot_a)
-    assert deployability_index.is_representative_not_deployable(snapshot_b)
-    assert not deployability_index.is_representative_not_deployable(snapshot_c)
+    assert not deployability_index.is_eligible_for_physical_layer(snapshot_a)
+    assert not deployability_index.is_eligible_for_physical_layer(snapshot_b)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_c)
 
 
 def test_deployability_index_uncategorized_forward_only_model(make_snapshot):
@@ -2686,8 +2686,8 @@ def test_deployability_index_uncategorized_forward_only_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
-    assert not deployability_index.is_representative_not_deployable(snapshot_a)
-    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_a)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_b)
 
 
 def test_deployability_index_auto_restatement_model(make_snapshot):
@@ -2718,8 +2718,8 @@ def test_deployability_index_auto_restatement_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
-    assert not deployability_index.is_representative_not_deployable(snapshot_a)
-    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_a)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_b)
 
 
 def test_deployability_index_categorized_forward_only_model(make_snapshot):
@@ -2750,8 +2750,8 @@ def test_deployability_index_categorized_forward_only_model(make_snapshot):
     assert not deployability_index.is_representative(snapshot_a)
     assert not deployability_index.is_representative(snapshot_b)
 
-    assert not deployability_index.is_representative_not_deployable(snapshot_a)
-    assert not deployability_index.is_representative_not_deployable(snapshot_b)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_a)
+    assert deployability_index.is_eligible_for_physical_layer(snapshot_b)
 
 
 def test_deployability_index_missing_parent(make_snapshot):