From 7e67cba7f416e74918f4a4ce8d59542e22ca6ffa Mon Sep 17 00:00:00 2001 From: David Cohen <16311507+dcohen24@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:00:30 -0500 Subject: [PATCH 1/5] fix(plan): account for stored parent changes when categorizing indirect snapshots When an indirectly modified snapshot had at least one direct parent that is new in the plan, only new parents were used to determine its category. A parent that already exists in state (categorized by an earlier plan, e.g. a breaking change applied to dev) but differs from the parent the previous child version was built from was ignored. A new metadata-only sibling could then make the child METADATA, reusing the old physical table and silently skipping the breaking change. Also collect categories from such stored parents by walking their version history back to the previous child's parent, reusing the walk of _get_orphaned_indirect_change_category through a shared helper. Unchanged parents and history at or before the baseline are ignored; a history that does not reach the baseline is treated as INDIRECT_BREAKING. Co-Authored-By: Claude Signed-off-by: David Cohen <16311507+dcohen24@users.noreply.github.com> --- sqlmesh/core/plan/builder.py | 78 ++++++++-- .../core/integration/test_change_scenarios.py | 139 ++++++++++++++++++ tests/core/test_plan.py | 122 +++++++++++++++ 3 files changed, 326 insertions(+), 13 deletions(-) diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index a6307a9ffd..b14d494295 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -743,7 +743,15 @@ def _categorize_snapshot( ): forward_only = True - if direct_parent_categories.intersection( + # Direct parents that already exist in state but changed relative to the previous version + # of this snapshot (eg. they were categorized in another plan) don't show up above. + stored_parent_categories = ( + self._get_stored_direct_parent_categories(snapshot) + if direct_parent_categories + else set() + ) + + if (direct_parent_categories | stored_parent_categories).intersection( {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING} ): snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING, forward_only) @@ -751,7 +759,9 @@ def _categorize_snapshot( snapshot.categorize_as( self._get_orphaned_indirect_change_category(snapshot), forward_only ) - elif all_upstream_categories == {SnapshotChangeCategory.METADATA}: + elif (all_upstream_categories | stored_parent_categories) == { + SnapshotChangeCategory.METADATA + }: snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only) else: snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING, forward_only) @@ -759,6 +769,55 @@ def _categorize_snapshot( # Metadata updated. snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only) + def _get_parent_categories_since( + self, parent: Snapshot, previous_parent_id: SnapshotId + ) -> t.Optional[t.Set[SnapshotChangeCategory]]: + """Returns the categories of all versions of the parent created after the previous parent, + including the current version. The previous parent itself is excluded. + + Returns None if the previous parent is not found in the parent's version history. + """ + categories: t.Set[SnapshotChangeCategory] = set() + for pv in reversed(parent.all_versions): + if pv.snapshot_id(parent.name) == previous_parent_id: + return categories + categories.add(t.cast(SnapshotChangeCategory, pv.change_category)) + return None + + def _get_stored_direct_parent_categories( + self, snapshot: Snapshot + ) -> t.Set[SnapshotChangeCategory]: + """Returns the change categories of direct parents that already exist in state (they are not new + in this plan) but changed since the previous version of the snapshot was built. + + For example, a parent was changed in a dev plan and is now combined with another parent + that is changed in the current plan. The change to the first parent is not visible to the + categorization of new parents, so it is recovered from the parent's version history. + """ + previous_snapshot = self._context_diff.modified_snapshots[snapshot.name][1] + previous_parent_ids = {p.name: p for p in previous_snapshot.parents} + + categories: t.Set[SnapshotChangeCategory] = set() + for p_id in snapshot.parents: + parent = self._context_diff.snapshots.get(p_id) + previous_parent_id = previous_parent_ids.get(p_id.name) + if not parent or self._is_new_snapshot(parent): + # Not stored, or categorized as part of this plan. + continue + if previous_parent_id is None or previous_parent_id == p_id: + # Not a parent of the previous version, or unchanged since then. + continue + + parent_categories = self._get_parent_categories_since(parent, previous_parent_id) + if parent_categories is None: + # The history doesn't reach back to the previous parent (it only keeps the latest + # versions), so we can't rule out a breaking change. A rebuild is safer than silently + # reusing stale data. + categories.add(SnapshotChangeCategory.INDIRECT_BREAKING) + else: + categories.update(parent_categories) + return categories + def _get_orphaned_indirect_change_category( self, indirect_snapshot: Snapshot ) -> SnapshotChangeCategory: @@ -790,17 +849,10 @@ def _get_orphaned_indirect_change_category( # so we can skip it continue - # Find the previous snapshot ID of the same parent in the historical chain - previous_parent_found = False - previous_parent_categories = set() - for pv in reversed(current_parent_snapshot.all_versions): - pv_snapshot_id = pv.snapshot_id(current_parent_snapshot.name) - if pv_snapshot_id == pevious_parent_snapshot_id: - previous_parent_found = True - break - previous_parent_categories.add(pv.change_category) - - if not previous_parent_found: + previous_parent_categories = self._get_parent_categories_since( + current_parent_snapshot, pevious_parent_snapshot_id + ) + if previous_parent_categories is None: # The previous parent is not in the historical chain so falling back to INDIRECT_BREAKING return SnapshotChangeCategory.INDIRECT_BREAKING diff --git a/tests/core/integration/test_change_scenarios.py b/tests/core/integration/test_change_scenarios.py index fb1762220f..4d237f148b 100644 --- a/tests/core/integration/test_change_scenarios.py +++ b/tests/core/integration/test_change_scenarios.py @@ -1515,3 +1515,142 @@ def test_creating_stage_for_first_batch_only(init_and_plan_context: t.Callable): assert ( context.engine_adapter.fetchone("SELECT COUNT(*) FROM test_schema.creating_counter")[0] == 1 ) + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +@pytest.mark.parametrize( + "stored_parent_category, new_parent_category, expected_child_category", + [ + ( + SnapshotChangeCategory.BREAKING, + SnapshotChangeCategory.METADATA, + SnapshotChangeCategory.INDIRECT_BREAKING, + ), + ( + SnapshotChangeCategory.BREAKING, + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.INDIRECT_BREAKING, + ), + ( + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.METADATA, + SnapshotChangeCategory.INDIRECT_NON_BREAKING, + ), + ( + SnapshotChangeCategory.METADATA, + SnapshotChangeCategory.METADATA, + SnapshotChangeCategory.METADATA, + ), + ], +) +def test_stored_changed_parent_with_new_parent( + init_and_plan_context: t.Callable, + stored_parent_category: SnapshotChangeCategory, # Stored in state (dev) but not deployed to prod + new_parent_category: SnapshotChangeCategory, # Introduced by the prod plan itself + expected_child_category: SnapshotChangeCategory, +): + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + initial_model_a = context.get_model("sushi.orders") + initial_model_b = context.get_model("sushi.items") + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # Make change A and deploy it to dev only. Prod keeps the previous version of the child. + context.upsert_model(initial_model_a.name, stamp="1") + parent_a = context.get_snapshot(initial_model_a.name, raise_if_missing=True) + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_a, stored_parent_category) + context.apply(plan_builder.build()) + + # Keep change A and make change B, creating a combination of parents not seen before. + context.upsert_model(initial_model_b.name, stamp="1") + parent_b = context.get_snapshot(initial_model_b.name, raise_if_missing=True) + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_b, new_parent_category) + plan = plan_builder.build() + + # Only B and its downstream snapshots are new in this plan. + assert parent_a.snapshot_id not in plan.context_diff.new_snapshots + assert parent_b.snapshot_id in plan.context_diff.new_snapshots + assert plan.snapshots[parent_a.snapshot_id].change_category == stored_parent_category + assert plan.snapshots[parent_b.snapshot_id].change_category == new_parent_category + + child_after = context.get_snapshot("sushi.order_items", raise_if_missing=True) + assert child_after.snapshot_id in plan.context_diff.new_snapshots + assert plan.snapshots[child_after.snapshot_id].change_category == expected_child_category + + # A breaking change to a parent requires a new version of the child, others reuse the old one. + child_requires_new_version = expected_child_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert (plan.snapshots[child_after.snapshot_id].version != child_before.version) == ( + child_requires_new_version + ) + assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) == ( + child_requires_new_version + ) + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +def test_unchanged_historic_breaking_parent_with_new_parent(init_and_plan_context: t.Callable): + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + initial_model_b = context.get_model("sushi.items") + + # Deploy a breaking change to the first parent to prod, including its children. + context.upsert_model("sushi.orders", stamp="1") + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice( + context.get_snapshot("sushi.orders", raise_if_missing=True), + SnapshotChangeCategory.BREAKING, + ) + context.apply(plan_builder.build()) + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # The first parent is unchanged relative to the deployed child, so its history must be ignored. + context.upsert_model(initial_model_b.name, stamp="1") + parent_b = context.get_snapshot(initial_model_b.name, raise_if_missing=True) + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_b, SnapshotChangeCategory.METADATA) + plan = plan_builder.build() + + child_after = context.get_snapshot("sushi.order_items", raise_if_missing=True) + assert child_after.snapshot_id in plan.context_diff.new_snapshots + assert ( + plan.snapshots[child_after.snapshot_id].change_category == SnapshotChangeCategory.METADATA + ) + assert plan.snapshots[child_after.snapshot_id].version == child_before.version + assert not any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +def test_stored_forward_only_parent_with_new_parent(init_and_plan_context: t.Callable): + """The forward-only flag of a stored parent doesn't make the child forward-only, the same way + it doesn't when both parents are new in the plan: the child is rebuilt.""" + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + initial_model_b = context.get_model("sushi.items") + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # Make sushi.orders a forward-only model and deploy a breaking change to it to dev only. + model = context.get_model("sushi.orders") + updated_model_kind = model.kind.copy(update={"forward_only": True}) + context.upsert_model(model.copy(update={"stamp": "1", "kind": updated_model_kind})) + parent_a = context.get_snapshot("sushi.orders", raise_if_missing=True) + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_a, SnapshotChangeCategory.BREAKING) + context.apply(plan_builder.build()) + assert context.get_snapshot("sushi.orders", raise_if_missing=True).is_forward_only + + context.upsert_model(initial_model_b.name, stamp="1") + parent_b = context.get_snapshot(initial_model_b.name, raise_if_missing=True) + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_b, SnapshotChangeCategory.METADATA) + plan = plan_builder.build() + + assert parent_a.snapshot_id not in plan.context_diff.new_snapshots + child_after = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id] + assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert not child_after.is_forward_only + assert child_after.version != child_before.version diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py index add85d2eec..cdb3b7fc11 100644 --- a/tests/core/test_plan.py +++ b/tests/core/test_plan.py @@ -4510,3 +4510,125 @@ def test_forward_only_indirect_change_to_materialized_view(make_snapshot): # Forward-only indirect changes to MVs should not always be classified as indirect breaking. # Instead, we want to preserve the standard categorization. assert snapshot_b_new.change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING + + +@pytest.mark.parametrize( + "stored_parent_categories, parent_baseline_in_history, expected_child_category", + [ + # The stored parent became breaking after the baseline, and its latest version is metadata. + ( + [ + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.BREAKING, + SnapshotChangeCategory.METADATA, + ], + True, + SnapshotChangeCategory.INDIRECT_BREAKING, + ), + # The baseline itself was breaking, but the previous child already includes that change. + ( + [SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.METADATA], + True, + SnapshotChangeCategory.METADATA, + ), + # Breaking changes at or before the baseline are already part of the previous child. + ( + [SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.NON_BREAKING], + True, + SnapshotChangeCategory.INDIRECT_NON_BREAKING, + ), + # A breaking change followed by a non-breaking one is still breaking after the baseline. + ( + [ + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.BREAKING, + SnapshotChangeCategory.NON_BREAKING, + ], + True, + SnapshotChangeCategory.INDIRECT_BREAKING, + ), + # The baseline can't be found in the parent's history, so the change can't be ruled out. + ( + [SnapshotChangeCategory.NON_BREAKING, SnapshotChangeCategory.METADATA], + False, + SnapshotChangeCategory.INDIRECT_BREAKING, + ), + ], +) +def test_indirect_change_with_stored_parent_history_and_new_parent( + make_snapshot, + stored_parent_categories: t.List[SnapshotChangeCategory], + parent_baseline_in_history: bool, + expected_child_category: SnapshotChangeCategory, +): + # The first category is assigned to the stored parent version that the previous child is built from + # (the baseline). The rest are assigned to versions stored after it. + def parent_snapshot(value: int, category: SnapshotChangeCategory): + snapshot = make_snapshot(SqlModel(name="a", query=parse_one(f"select {value} as col_a"))) + snapshot.categorize_as(category) + return snapshot + + parent_versions = [ + parent_snapshot(i, category) for i, category in enumerate(stored_parent_categories) + ] + for i, current in enumerate(parent_versions[1:], start=1): + if i == 1 and not parent_baseline_in_history: + continue + current.previous_versions = parent_versions[i - 1].all_versions + baseline_parent, stored_parent = parent_versions[0], parent_versions[-1] + + parent_b_old = make_snapshot(SqlModel(name="b", query=parse_one("select 1 as col_b"))) + parent_b_old.categorize_as(SnapshotChangeCategory.BREAKING) + parent_b_new = make_snapshot( + SqlModel(name="b", query=parse_one("select 1 as col_b"), owner="new_owner") + ) + parent_b_new.previous_versions = parent_b_old.all_versions + + child_model = SqlModel(name="c", query=parse_one("select col_a, col_b from a, b")) + child_old = make_snapshot( + child_model, + nodes={'"a"': baseline_parent.model, '"b"': parent_b_old.model}, + ) + child_old.categorize_as(SnapshotChangeCategory.BREAKING) + child_new = make_snapshot( + child_model, + nodes={'"a"': stored_parent.model, '"b"': parent_b_new.model}, + ) + child_new.previous_versions = child_old.all_versions + + context_diff = ContextDiff( + environment="test_environment", + is_new_environment=True, + is_unfinalized_environment=False, + normalize_environment_name=True, + create_from="prod", + create_from_env_exists=True, + added=set(), + removed_snapshots={}, + modified_snapshots={ + stored_parent.name: (stored_parent, baseline_parent), + parent_b_new.name: (parent_b_new, parent_b_old), + child_new.name: (child_new, child_old), + }, + snapshots={ + stored_parent.snapshot_id: stored_parent, + parent_b_new.snapshot_id: parent_b_new, + child_new.snapshot_id: child_new, + }, + # The stored parent was categorized in an earlier plan, so it's not new in this one. + new_snapshots={ + parent_b_new.snapshot_id: parent_b_new, + child_new.snapshot_id: child_new, + }, + previous_plan_id=None, + previously_promoted_snapshot_ids=set(), + previous_finalized_snapshots=None, + previous_gateway_managed_virtual_layer=False, + gateway_managed_virtual_layer=False, + environment_statements=[], + ) + + PlanBuilder(context_diff, forward_only=False).build() + + assert parent_b_new.change_category == SnapshotChangeCategory.METADATA + assert child_new.change_category == expected_child_category From bbf0a6142eaa0152ce5b724ead0bc26c8039a0cd Mon Sep 17 00:00:00 2001 From: David Cohen <16311507+dcohen24@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:24:11 -0500 Subject: [PATCH 2/5] fix(plan): include stored parents in the forward-only decision A stored parent with a non-forward-only breaking change combined with a new forward-only parent left the child flagged forward-only, since only new parents were considered. The child was categorized INDIRECT_BREAKING but reused its previous physical version, skipping the required rebuild. Add the forward-only flags of stored parents that changed since the previous child version to the forward-only decision, and cover both forward-only parent combinations with tests. Co-Authored-By: Claude Signed-off-by: David Cohen <16311507+dcohen24@users.noreply.github.com> --- sqlmesh/core/plan/builder.py | 32 +++++++++------- .../core/integration/test_change_scenarios.py | 38 +++++++++++++++++++ 2 files changed, 57 insertions(+), 13 deletions(-) diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index b14d494295..e64c1ddb36 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -738,19 +738,20 @@ def _categorize_snapshot( if p_id in snapshot.parents: direct_parent_categories.add(parent.change_category) + # Direct parents that already exist in state but changed relative to the previous version + # of this snapshot (eg. they were categorized in another plan) don't show up above. + stored_parent_categories: t.Set[SnapshotChangeCategory] = set() + if direct_parent_categories: + stored_parent_categories, stored_parent_forward_only = ( + self._get_stored_direct_parent_changes(snapshot) + ) + all_upstream_forward_only |= stored_parent_forward_only + if all_upstream_forward_only == {True} or ( snapshot.is_model and snapshot.model.forward_only ): forward_only = True - # Direct parents that already exist in state but changed relative to the previous version - # of this snapshot (eg. they were categorized in another plan) don't show up above. - stored_parent_categories = ( - self._get_stored_direct_parent_categories(snapshot) - if direct_parent_categories - else set() - ) - if (direct_parent_categories | stored_parent_categories).intersection( {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING} ): @@ -784,11 +785,12 @@ def _get_parent_categories_since( categories.add(t.cast(SnapshotChangeCategory, pv.change_category)) return None - def _get_stored_direct_parent_categories( + def _get_stored_direct_parent_changes( self, snapshot: Snapshot - ) -> t.Set[SnapshotChangeCategory]: - """Returns the change categories of direct parents that already exist in state (they are not new - in this plan) but changed since the previous version of the snapshot was built. + ) -> t.Tuple[t.Set[SnapshotChangeCategory], t.Set[bool]]: + """Returns the change categories and forward-only flags of direct parents that already exist in + state (they are not new in this plan) but changed since the previous version of the snapshot + was built. For example, a parent was changed in a dev plan and is now combined with another parent that is changed in the current plan. The change to the first parent is not visible to the @@ -798,6 +800,7 @@ def _get_stored_direct_parent_categories( previous_parent_ids = {p.name: p for p in previous_snapshot.parents} categories: t.Set[SnapshotChangeCategory] = set() + forward_only: t.Set[bool] = set() for p_id in snapshot.parents: parent = self._context_diff.snapshots.get(p_id) previous_parent_id = previous_parent_ids.get(p_id.name) @@ -808,6 +811,9 @@ def _get_stored_direct_parent_categories( # Not a parent of the previous version, or unchanged since then. continue + # The change is applied the way the parent was categorized, so it counts toward + # whether the snapshot can skip a rebuild, the same as the new parents. + forward_only.add(parent.is_forward_only) parent_categories = self._get_parent_categories_since(parent, previous_parent_id) if parent_categories is None: # The history doesn't reach back to the previous parent (it only keeps the latest @@ -816,7 +822,7 @@ def _get_stored_direct_parent_categories( categories.add(SnapshotChangeCategory.INDIRECT_BREAKING) else: categories.update(parent_categories) - return categories + return categories, forward_only def _get_orphaned_indirect_change_category( self, indirect_snapshot: Snapshot diff --git a/tests/core/integration/test_change_scenarios.py b/tests/core/integration/test_change_scenarios.py index 4d237f148b..a24c39706a 100644 --- a/tests/core/integration/test_change_scenarios.py +++ b/tests/core/integration/test_change_scenarios.py @@ -1654,3 +1654,41 @@ def test_stored_forward_only_parent_with_new_parent(init_and_plan_context: t.Cal assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING assert not child_after.is_forward_only assert child_after.version != child_before.version + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +def test_stored_breaking_parent_with_new_forward_only_parent(init_and_plan_context: t.Callable): + """A forward-only change to a new parent must not make the child forward-only when a stored + parent also changed in a non-forward-only way, otherwise the child skips the required rebuild.""" + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # Make a breaking, not forward-only, change to sushi.orders and deploy it to dev only. + context.upsert_model("sushi.orders", stamp="1") + parent_a = context.get_snapshot("sushi.orders", raise_if_missing=True) + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_a, SnapshotChangeCategory.BREAKING) + context.apply(plan_builder.build()) + assert not context.get_snapshot("sushi.orders", raise_if_missing=True).is_forward_only + + # Make a forward-only change to sushi.items, which is new in the prod plan. + model = context.get_model("sushi.items") + updated_model_kind = model.kind.copy(update={"forward_only": True}) + context.upsert_model(model.copy(update={"stamp": "1", "kind": updated_model_kind})) + parent_b = context.get_snapshot("sushi.items", raise_if_missing=True) + plan = context.plan_builder( + "prod", + skip_tests=True, + enable_preview=False, + categorizer_config=CategorizerConfig.all_full(), + ).build() + + assert parent_a.snapshot_id not in plan.context_diff.new_snapshots + assert plan.snapshots[parent_b.snapshot_id].is_forward_only + child_after = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id] + assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert not child_after.is_forward_only + assert child_after.version != child_before.version + assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) From 0162ab21a94f36d66036775394b5d234e5deaafa Mon Sep 17 00:00:00 2001 From: David Cohen <16311507+dcohen24@users.noreply.github.com> Date: Thu, 12 Nov 2026 21:11:32 -0500 Subject: [PATCH 3/5] fix(plan): don't let metadata or non-breaking changes veto forward-only for stored parents When a stored parent has changed since the previous child version, decide whether the child is forward-only only from the changes that require downstream history to be recomputed. New parents categorized METADATA, NON_BREAKING or INDIRECT_NON_BREAKING no longer count against a forward-only stored parent, and stored parents only count when their history contains a breaking change. A stored parent whose history can't be traced counts as not forward-only. Plans without a changed stored parent decide exactly as before. Extend the forward-only test to cover metadata, non-breaking and breaking new parents. Co-Authored-By: Claude Signed-off-by: David Cohen <16311507+dcohen24@users.noreply.github.com> --- sqlmesh/core/plan/builder.py | 30 ++++++++++++++++--- .../core/integration/test_change_scenarios.py | 29 ++++++++++++++---- 2 files changed, 49 insertions(+), 10 deletions(-) diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index e64c1ddb36..2be4ad56b9 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -52,6 +52,14 @@ logger = logging.getLogger(__name__) +# Changes in these categories don't require the history of downstream snapshots to be recomputed. +DOWNSTREAM_REBUILD_NOT_REQUIRED_CATEGORIES = { + SnapshotChangeCategory.METADATA, + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.INDIRECT_NON_BREAKING, +} + + class PlanBuilder: """Plan Builder constructs a Plan based on user choices for how they want to backfill, preview, etc. their changes. @@ -726,6 +734,7 @@ def _categorize_snapshot( return all_upstream_forward_only = set() + data_changing_upstream_forward_only = set() all_upstream_categories = set() direct_parent_categories = set() @@ -735,6 +744,8 @@ def _categorize_snapshot( if parent and self._is_new_snapshot(parent): all_upstream_categories.add(parent.change_category) all_upstream_forward_only.add(parent.is_forward_only) + if parent.change_category not in DOWNSTREAM_REBUILD_NOT_REQUIRED_CATEGORIES: + data_changing_upstream_forward_only.add(parent.is_forward_only) if p_id in snapshot.parents: direct_parent_categories.add(parent.change_category) @@ -745,7 +756,14 @@ def _categorize_snapshot( stored_parent_categories, stored_parent_forward_only = ( self._get_stored_direct_parent_changes(snapshot) ) - all_upstream_forward_only |= stored_parent_forward_only + + if stored_parent_categories: + # A parent that changed in another plan is part of this change too. Whether the snapshot + # can skip a rebuild now depends on the changes that require downstream history to be + # recomputed: a metadata or non-breaking change doesn't, so it can't veto forward-only. + all_upstream_forward_only = ( + data_changing_upstream_forward_only | stored_parent_forward_only + ) if all_upstream_forward_only == {True} or ( snapshot.is_model and snapshot.model.forward_only @@ -811,17 +829,21 @@ def _get_stored_direct_parent_changes( # Not a parent of the previous version, or unchanged since then. continue - # The change is applied the way the parent was categorized, so it counts toward - # whether the snapshot can skip a rebuild, the same as the new parents. - forward_only.add(parent.is_forward_only) parent_categories = self._get_parent_categories_since(parent, previous_parent_id) if parent_categories is None: # The history doesn't reach back to the previous parent (it only keeps the latest # versions), so we can't rule out a breaking change. A rebuild is safer than silently # reusing stale data. categories.add(SnapshotChangeCategory.INDIRECT_BREAKING) + forward_only.add(False) else: categories.update(parent_categories) + if parent_categories.intersection( + {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING} + ): + # The parent's history changed in a way that requires downstream history to be + # recomputed, unless that change was applied forward-only. + forward_only.add(parent.is_forward_only) return categories, forward_only def _get_orphaned_indirect_change_category( diff --git a/tests/core/integration/test_change_scenarios.py b/tests/core/integration/test_change_scenarios.py index a24c39706a..a6b9d389b5 100644 --- a/tests/core/integration/test_change_scenarios.py +++ b/tests/core/integration/test_change_scenarios.py @@ -1624,9 +1624,22 @@ def test_unchanged_historic_breaking_parent_with_new_parent(init_and_plan_contex @time_machine.travel("2023-01-08 15:00:00 UTC") -def test_stored_forward_only_parent_with_new_parent(init_and_plan_context: t.Callable): - """The forward-only flag of a stored parent doesn't make the child forward-only, the same way - it doesn't when both parents are new in the plan: the child is rebuilt.""" +@pytest.mark.parametrize( + "new_parent_category, expected_forward_only", + [ + # These changes don't require downstream history to be recomputed, so they don't affect + # the forward-only decision. + (SnapshotChangeCategory.METADATA, True), + (SnapshotChangeCategory.NON_BREAKING, True), + # A breaking change that is not forward-only changes the history of the new parent. + (SnapshotChangeCategory.BREAKING, False), + ], +) +def test_stored_forward_only_parent_with_new_parent( + init_and_plan_context: t.Callable, + new_parent_category: SnapshotChangeCategory, + expected_forward_only: bool, +): context, plan = init_and_plan_context("examples/sushi") context.apply(plan) @@ -1646,14 +1659,18 @@ def test_stored_forward_only_parent_with_new_parent(init_and_plan_context: t.Cal context.upsert_model(initial_model_b.name, stamp="1") parent_b = context.get_snapshot(initial_model_b.name, raise_if_missing=True) plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) - plan_builder.set_choice(parent_b, SnapshotChangeCategory.METADATA) + plan_builder.set_choice(parent_b, new_parent_category) plan = plan_builder.build() assert parent_a.snapshot_id not in plan.context_diff.new_snapshots child_after = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id] assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING - assert not child_after.is_forward_only - assert child_after.version != child_before.version + assert child_after.is_forward_only == expected_forward_only + # A forward-only child keeps its previous version, otherwise it's rebuilt. + assert (child_after.version == child_before.version) == expected_forward_only + assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) == ( + not expected_forward_only + ) @time_machine.travel("2023-01-08 15:00:00 UTC") From 19c72ed27e496f2bab9637b13f1bcc367f7cc875 Mon Sep 17 00:00:00 2001 From: David Cohen <16311507+dcohen24@users.noreply.github.com> Date: Thu, 12 Nov 2026 21:53:48 -0500 Subject: [PATCH 4/5] fix(plan): decide forward-only from every breaking version of a stored parent A stored parent only voted with the forward-only flag of its latest version. If an earlier breaking version since the previous child was not forward-only and the latest one was, the child was flagged forward-only and kept its old version, so the earlier change was never applied to it. A stored parent now votes forward-only only if every breaking version since the previous child was applied forward-only. Past versions don't record the flag, so it is inferred from whether the version reused the one before it. That inference is only used when virtual_environment_mode is FULL, since other modes don't derive versions from the fingerprint. Co-Authored-By: Claude Signed-off-by: David Cohen <16311507+dcohen24@users.noreply.github.com> --- sqlmesh/core/plan/builder.py | 65 +++++++++---- .../core/integration/test_change_scenarios.py | 44 +++++++++ tests/core/test_plan.py | 92 +++++++++++++++++++ 3 files changed, 183 insertions(+), 18 deletions(-) diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index 2be4ad56b9..bffa89eb5b 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -35,7 +35,7 @@ SnapshotChangeCategory, ) from sqlmesh.core.snapshot.categorizer import categorize_change -from sqlmesh.core.snapshot.definition import Interval, SnapshotId +from sqlmesh.core.snapshot.definition import Interval, SnapshotDataVersion, SnapshotId from sqlmesh.utils import columns_to_types_all_known, random_id from sqlmesh.utils.dag import DAG from sqlmesh.utils.date import ( @@ -788,21 +788,41 @@ def _categorize_snapshot( # Metadata updated. snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only) - def _get_parent_categories_since( + def _get_parent_versions_since( self, parent: Snapshot, previous_parent_id: SnapshotId - ) -> t.Optional[t.Set[SnapshotChangeCategory]]: - """Returns the categories of all versions of the parent created after the previous parent, - including the current version. The previous parent itself is excluded. + ) -> t.Optional[t.List[SnapshotDataVersion]]: + """Returns all versions of the parent created after the previous parent, including the current + version. The previous parent itself is excluded. Returns None if the previous parent is not found in the parent's version history. """ - categories: t.Set[SnapshotChangeCategory] = set() + versions: t.List[SnapshotDataVersion] = [] for pv in reversed(parent.all_versions): if pv.snapshot_id(parent.name) == previous_parent_id: - return categories - categories.add(t.cast(SnapshotChangeCategory, pv.change_category)) + return versions + versions.append(pv) return None + def _get_parent_categories_since( + self, parent: Snapshot, previous_parent_id: SnapshotId + ) -> t.Optional[t.Set[SnapshotChangeCategory]]: + """Returns the categories of the versions returned by `_get_parent_versions_since`.""" + versions = self._get_parent_versions_since(parent, previous_parent_id) + if versions is None: + return None + return {t.cast(SnapshotChangeCategory, v.change_category) for v in versions} + + @staticmethod + def _was_applied_forward_only(version: SnapshotDataVersion) -> bool: + """Returns True if the version reused the table of the version before it. + + Past versions don't record whether they were forward-only, but a forward-only version reuses the + previous version, while any other breaking version gets a new one. This doesn't hold when the + version is not derived from the fingerprint (virtual environments that are not fully enabled), + so these are never considered forward-only. + """ + return version.virtual_environment_mode.is_full and not version.is_new_version + def _get_stored_direct_parent_changes( self, snapshot: Snapshot ) -> t.Tuple[t.Set[SnapshotChangeCategory], t.Set[bool]]: @@ -829,21 +849,30 @@ def _get_stored_direct_parent_changes( # Not a parent of the previous version, or unchanged since then. continue - parent_categories = self._get_parent_categories_since(parent, previous_parent_id) - if parent_categories is None: + parent_versions = self._get_parent_versions_since(parent, previous_parent_id) + if parent_versions is None: # The history doesn't reach back to the previous parent (it only keeps the latest # versions), so we can't rule out a breaking change. A rebuild is safer than silently # reusing stale data. categories.add(SnapshotChangeCategory.INDIRECT_BREAKING) forward_only.add(False) - else: - categories.update(parent_categories) - if parent_categories.intersection( - {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING} - ): - # The parent's history changed in a way that requires downstream history to be - # recomputed, unless that change was applied forward-only. - forward_only.add(parent.is_forward_only) + continue + + categories.update( + t.cast(SnapshotChangeCategory, v.change_category) for v in parent_versions + ) + + # Every breaking version of the parent since the previous child requires downstream history to + # be recomputed, unless it was applied forward-only. The parent's latest version being + # forward-only is not enough, an earlier version may not have been. + breaking_versions = [ + v + for v in parent_versions + if v.change_category + in (SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING) + ] + if breaking_versions: + forward_only.add(all(self._was_applied_forward_only(v) for v in breaking_versions)) return categories, forward_only def _get_orphaned_indirect_change_category( diff --git a/tests/core/integration/test_change_scenarios.py b/tests/core/integration/test_change_scenarios.py index a6b9d389b5..eb8a999920 100644 --- a/tests/core/integration/test_change_scenarios.py +++ b/tests/core/integration/test_change_scenarios.py @@ -1709,3 +1709,47 @@ def test_stored_breaking_parent_with_new_forward_only_parent(init_and_plan_conte assert not child_after.is_forward_only assert child_after.version != child_before.version assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +def test_stored_parent_history_with_earlier_non_forward_only_change( + init_and_plan_context: t.Callable, +): + """Only the latest version of the stored parent is forward-only. An earlier breaking version + that is not forward-only was never applied to the child, so the child has to be rebuilt.""" + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # First, a breaking change that is not forward-only is deployed to dev only. + context.upsert_model("sushi.orders", stamp="1") + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice( + context.get_snapshot("sushi.orders", raise_if_missing=True), SnapshotChangeCategory.BREAKING + ) + context.apply(plan_builder.build()) + + # Then a forward-only breaking change to the same model is deployed to dev only. + model = context.get_model("sushi.orders") + updated_model_kind = model.kind.copy(update={"forward_only": True}) + context.upsert_model(model.copy(update={"stamp": "2", "kind": updated_model_kind})) + parent_a = context.get_snapshot("sushi.orders", raise_if_missing=True) + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_a, SnapshotChangeCategory.BREAKING) + context.apply(plan_builder.build()) + assert context.get_snapshot("sushi.orders", raise_if_missing=True).is_forward_only + + # The prod plan only creates a metadata change to the other parent. + context.upsert_model("sushi.items", stamp="1") + parent_b = context.get_snapshot("sushi.items", raise_if_missing=True) + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice(parent_b, SnapshotChangeCategory.METADATA) + plan = plan_builder.build() + + assert parent_a.snapshot_id not in plan.context_diff.new_snapshots + child_after = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id] + assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert not child_after.is_forward_only + assert child_after.version != child_before.version + assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py index cdb3b7fc11..434ecf10cd 100644 --- a/tests/core/test_plan.py +++ b/tests/core/test_plan.py @@ -5,6 +5,7 @@ import pytest +from sqlmesh.core.config.common import VirtualEnvironmentMode from sqlmesh.core.console import TerminalConsole from sqlmesh.utils.metaprogramming import Executable from tests.core.test_table_diff import create_test_console @@ -4632,3 +4633,94 @@ def parent_snapshot(value: int, category: SnapshotChangeCategory): assert parent_b_new.change_category == SnapshotChangeCategory.METADATA assert child_new.change_category == expected_child_category + + +@pytest.mark.parametrize( + "virtual_environment_mode, latest_version_forward_only, expected_forward_only", + [ + # The latest breaking version of the stored parent reused the previous version. + (VirtualEnvironmentMode.FULL, True, True), + # The latest breaking version of the stored parent got a new version. + (VirtualEnvironmentMode.FULL, False, False), + # The versions don't depend on the fingerprint, so they can't be told apart from forward-only ones. + (VirtualEnvironmentMode.DEV_ONLY, True, False), + ], +) +def test_indirect_change_forward_only_with_stored_parent_history( + make_snapshot, + virtual_environment_mode: VirtualEnvironmentMode, + latest_version_forward_only: bool, + expected_forward_only: bool, +): + def parent_snapshot(value: int): + return make_snapshot( + SqlModel( + name="a", + query=parse_one(f"select {value} as col_a"), + virtual_environment_mode=virtual_environment_mode, + ) + ) + + parent_baseline = parent_snapshot(0) + parent_baseline.categorize_as(SnapshotChangeCategory.BREAKING) + stored_parent = parent_snapshot(1) + stored_parent.previous_versions = parent_baseline.all_versions + stored_parent.categorize_as( + SnapshotChangeCategory.BREAKING, forward_only=latest_version_forward_only + ) + + parent_b_old = make_snapshot(SqlModel(name="b", query=parse_one("select 1 as col_b"))) + parent_b_old.categorize_as(SnapshotChangeCategory.BREAKING) + parent_b_new = make_snapshot( + SqlModel(name="b", query=parse_one("select 1 as col_b"), owner="new_owner") + ) + parent_b_new.previous_versions = parent_b_old.all_versions + + child_model = SqlModel(name="c", query=parse_one("select col_a, col_b from a, b")) + child_old = make_snapshot( + child_model, + nodes={'"a"': parent_baseline.model, '"b"': parent_b_old.model}, + ) + child_old.categorize_as(SnapshotChangeCategory.BREAKING) + child_new = make_snapshot( + child_model, + nodes={'"a"': stored_parent.model, '"b"': parent_b_new.model}, + ) + child_new.previous_versions = child_old.all_versions + + context_diff = ContextDiff( + environment="test_environment", + is_new_environment=True, + is_unfinalized_environment=False, + normalize_environment_name=True, + create_from="prod", + create_from_env_exists=True, + added=set(), + removed_snapshots={}, + modified_snapshots={ + stored_parent.name: (stored_parent, parent_baseline), + parent_b_new.name: (parent_b_new, parent_b_old), + child_new.name: (child_new, child_old), + }, + snapshots={ + stored_parent.snapshot_id: stored_parent, + parent_b_new.snapshot_id: parent_b_new, + child_new.snapshot_id: child_new, + }, + # The stored parent was categorized in an earlier plan, so it's not new in this one. + new_snapshots={ + parent_b_new.snapshot_id: parent_b_new, + child_new.snapshot_id: child_new, + }, + previous_plan_id=None, + previously_promoted_snapshot_ids=set(), + previous_finalized_snapshots=None, + previous_gateway_managed_virtual_layer=False, + gateway_managed_virtual_layer=False, + environment_statements=[], + ) + + PlanBuilder(context_diff, forward_only=False).build() + + assert child_new.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert child_new.is_forward_only == expected_forward_only From 64a96184420d8e1acf37a552c873f9722ef3914b Mon Sep 17 00:00:00 2001 From: David Cohen <16311507+dcohen24@users.noreply.github.com> Date: Fri, 13 Nov 2026 08:02:00 -0500 Subject: [PATCH 5/5] fix(plan): read the forward-only flag of stored parent versions from state The forward-only decision for a child with a stored parent inferred whether earlier versions of that parent were forward-only from their version and fingerprint. That is wrong for pinned versions, which are not derived from the fingerprint: a pinned parent switched from DEV_ONLY to FULL was a rebuild, but looked forward-only, so its child reused the old table. ContextDiff now loads the stored snapshot of each earlier version of a stored parent after the baseline and records its forward-only flag. The plan builder uses that flag, and the parent's own flag for its latest version. A version that can't be found in state is not considered forward-only. Categories that are not assigned today (none, or the deprecated FORWARD_ONLY) are treated as breaking. Co-Authored-By: Claude Signed-off-by: David Cohen <16311507+dcohen24@users.noreply.github.com> --- sqlmesh/core/context_diff.py | 47 ++++++++++++++ sqlmesh/core/plan/builder.py | 45 ++++++++++---- .../core/integration/test_change_scenarios.py | 61 +++++++++++++++++++ tests/core/test_plan.py | 47 ++++++++------ 4 files changed, 169 insertions(+), 31 deletions(-) diff --git a/sqlmesh/core/context_diff.py b/sqlmesh/core/context_diff.py index 047e58609a..4eeb4e4ed8 100644 --- a/sqlmesh/core/context_diff.py +++ b/sqlmesh/core/context_diff.py @@ -88,6 +88,9 @@ class ContextDiff(PydanticModel): """Environment statements.""" diff_rendered: bool = False """Whether the diff should compare raw vs rendered models""" + stored_parent_forward_only: t.Dict[SnapshotId, bool] = {} + """Whether the earlier versions of the stored parents of modified snapshots were forward-only. + A past version doesn't record this, so it is read from the snapshot stored for that version.""" @classmethod def create( @@ -233,6 +236,9 @@ def create( return ContextDiff( environment=environment, + stored_parent_forward_only=_get_stored_parent_forward_only( + state_reader, merged_snapshots, new_snapshots, modified_snapshots + ), is_new_environment=is_new_environment, is_unfinalized_environment=bool(env and not env.finalized_ts), normalize_environment_name=is_new_environment or bool(env and env.normalize_name), @@ -491,6 +497,47 @@ def text_diff(self, name: str) -> str: return "" +def _get_stored_parent_forward_only( + state_reader: StateReader, + snapshots: t.Dict[SnapshotId, Snapshot], + new_snapshots: t.Dict[SnapshotId, Snapshot], + modified_snapshots: t.Dict[str, t.Tuple[Snapshot, Snapshot]], +) -> t.Dict[SnapshotId, bool]: + """Returns whether the past versions of stored parents were forward-only. + + A stored parent already exists in state and is a parent of a new modified snapshot, but it has changed + since that snapshot's previous version. Only the versions created after that change are returned, + and those that can't be found in state are left out. + """ + version_ids: t.Set[SnapshotId] = set() + for snapshot, previous_snapshot in modified_snapshots.values(): + if snapshot.snapshot_id not in new_snapshots: + continue + previous_parent_ids = {p.name: p for p in previous_snapshot.parents} + for parent_id in snapshot.parents: + parent = snapshots.get(parent_id) + previous_parent_id = previous_parent_ids.get(parent_id.name) + if ( + not parent + or parent_id in new_snapshots + or previous_parent_id is None + or previous_parent_id == parent_id + ): + continue + for version in reversed(parent.previous_versions): + version_id = version.snapshot_id(parent.name) + if version_id == previous_parent_id: + break + version_ids.add(version_id) + + if not version_ids: + return {} + return { + snapshot_id: snapshot.is_forward_only + for snapshot_id, snapshot in state_reader.get_snapshots(version_ids).items() + } + + def _build_requirements( provided_requirements: t.Dict[str, str], excluded_requirements: t.Set[str], diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py index bffa89eb5b..a13362c4c3 100644 --- a/sqlmesh/core/plan/builder.py +++ b/sqlmesh/core/plan/builder.py @@ -813,15 +813,33 @@ def _get_parent_categories_since( return {t.cast(SnapshotChangeCategory, v.change_category) for v in versions} @staticmethod - def _was_applied_forward_only(version: SnapshotDataVersion) -> bool: - """Returns True if the version reused the table of the version before it. + def _get_known_category(version: SnapshotDataVersion) -> SnapshotChangeCategory: + """Returns the category of the version, or BREAKING if it isn't one that is assigned today. - Past versions don't record whether they were forward-only, but a forward-only version reuses the - previous version, while any other breaking version gets a new one. This doesn't hold when the - version is not derived from the fingerprint (virtual environments that are not fully enabled), - so these are never considered forward-only. + A missing category or the deprecated FORWARD_ONLY category may still be found in older state. Both + changed data, so they are treated as breaking, and whether they were forward-only is read from the + stored snapshot like for any other breaking version. """ - return version.virtual_environment_mode.is_full and not version.is_new_version + if version.change_category in ( + SnapshotChangeCategory.BREAKING, + SnapshotChangeCategory.INDIRECT_BREAKING, + SnapshotChangeCategory.NON_BREAKING, + SnapshotChangeCategory.INDIRECT_NON_BREAKING, + SnapshotChangeCategory.METADATA, + ): + return version.change_category + return SnapshotChangeCategory.BREAKING + + def _was_applied_forward_only(self, parent: Snapshot, version: SnapshotDataVersion) -> bool: + """Returns True if the given version of the parent was applied forward-only. + + A past version doesn't record this, so it is read from the snapshot stored for that version. + If that snapshot can't be found, we can't tell, so the version is not considered forward-only. + """ + version_id = version.snapshot_id(parent.name) + if version_id == parent.snapshot_id: + return parent.is_forward_only + return self._context_diff.stored_parent_forward_only.get(version_id, False) def _get_stored_direct_parent_changes( self, snapshot: Snapshot @@ -858,21 +876,22 @@ def _get_stored_direct_parent_changes( forward_only.add(False) continue - categories.update( - t.cast(SnapshotChangeCategory, v.change_category) for v in parent_versions - ) + version_categories = [(v, self._get_known_category(v)) for v in parent_versions] + categories.update(category for _, category in version_categories) # Every breaking version of the parent since the previous child requires downstream history to # be recomputed, unless it was applied forward-only. The parent's latest version being # forward-only is not enough, an earlier version may not have been. breaking_versions = [ v - for v in parent_versions - if v.change_category + for v, category in version_categories + if category in (SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING) ] if breaking_versions: - forward_only.add(all(self._was_applied_forward_only(v) for v in breaking_versions)) + forward_only.add( + all(self._was_applied_forward_only(parent, v) for v in breaking_versions) + ) return categories, forward_only def _get_orphaned_indirect_change_category( diff --git a/tests/core/integration/test_change_scenarios.py b/tests/core/integration/test_change_scenarios.py index eb8a999920..4afba4d5c2 100644 --- a/tests/core/integration/test_change_scenarios.py +++ b/tests/core/integration/test_change_scenarios.py @@ -24,6 +24,7 @@ ) from sqlmesh.core.context import Context from sqlmesh.core.config.categorizer import CategorizerConfig +from sqlmesh.core.config.common import VirtualEnvironmentMode from sqlmesh.core.model import ( FullKind, ModelKind, @@ -1753,3 +1754,63 @@ def test_stored_parent_history_with_earlier_non_forward_only_change( assert not child_after.is_forward_only assert child_after.version != child_before.version assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) + + +@time_machine.travel("2023-01-08 15:00:00 UTC") +@pytest.mark.parametrize("with_new_parent", [False, True]) +def test_stored_pinned_parent_switched_to_full_virtual_environment_mode( + init_and_plan_context: t.Callable, with_new_parent: bool +): + """A pinned version is not derived from the fingerprint, so it can't be used to tell whether the + stored parent was applied forward-only. Switching the virtual environment mode to FULL requires + a rebuild, adding a metadata-only parent change must not change that outcome.""" + context, plan = init_and_plan_context("examples/sushi") + context.apply(plan) + + # Pin the version of sushi.orders in a DEV_ONLY virtual environment and deploy it to prod. + model = context.get_model("sushi.orders") + context.upsert_model( + model.copy( + update={ + "stamp": "1", + "kind": model.kind.copy(update={"forward_only": True}), + "physical_version": "pinned", + "virtual_environment_mode": VirtualEnvironmentMode.DEV_ONLY, + } + ) + ) + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + plan_builder.set_choice( + context.get_snapshot("sushi.orders", raise_if_missing=True), SnapshotChangeCategory.BREAKING + ) + context.apply(plan_builder.build()) + child_before = context.get_snapshot("sushi.order_items", raise_if_missing=True) + + # Switch the model to FULL virtual environment mode in dev only. + model = context.get_model("sushi.orders") + context.upsert_model( + model.copy(update={"stamp": "2", "virtual_environment_mode": VirtualEnvironmentMode.FULL}) + ) + plan_builder = context.plan_builder("dev", skip_tests=True, enable_preview=False) + plan_builder.set_choice( + context.get_snapshot("sushi.orders", raise_if_missing=True), SnapshotChangeCategory.BREAKING + ) + context.apply(plan_builder.build()) + parent_a = context.get_snapshot("sushi.orders", raise_if_missing=True) + assert not parent_a.is_forward_only + + if with_new_parent: + context.upsert_model("sushi.items", stamp="1") + plan_builder = context.plan_builder("prod", skip_tests=True, enable_preview=False) + if with_new_parent: + plan_builder.set_choice( + context.get_snapshot("sushi.items", raise_if_missing=True), + SnapshotChangeCategory.METADATA, + ) + plan = plan_builder.build() + + child_after = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id] + assert child_after.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert not child_after.is_forward_only + assert child_after.version != child_before.version + assert any(i.snapshot_id == child_after.snapshot_id for i in plan.missing_intervals) diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py index 434ecf10cd..8428e9130c 100644 --- a/tests/core/test_plan.py +++ b/tests/core/test_plan.py @@ -5,7 +5,6 @@ import pytest -from sqlmesh.core.config.common import VirtualEnvironmentMode from sqlmesh.core.console import TerminalConsole from sqlmesh.utils.metaprogramming import Executable from tests.core.test_table_diff import create_test_console @@ -4636,35 +4635,42 @@ def parent_snapshot(value: int, category: SnapshotChangeCategory): @pytest.mark.parametrize( - "virtual_environment_mode, latest_version_forward_only, expected_forward_only", + "earlier_version_flag, earlier_version_category, latest_version_forward_only, expected_forward_only", [ - # The latest breaking version of the stored parent reused the previous version. - (VirtualEnvironmentMode.FULL, True, True), - # The latest breaking version of the stored parent got a new version. - (VirtualEnvironmentMode.FULL, False, False), - # The versions don't depend on the fingerprint, so they can't be told apart from forward-only ones. - (VirtualEnvironmentMode.DEV_ONLY, True, False), + # Both breaking versions of the stored parent were forward-only. + (True, SnapshotChangeCategory.BREAKING, True, True), + # The latest version was not forward-only. + (True, SnapshotChangeCategory.BREAKING, False, False), + # An earlier version was not forward-only. + (False, SnapshotChangeCategory.BREAKING, True, False), + # An earlier version can't be found in state, so we can't tell how it was applied. + (None, SnapshotChangeCategory.BREAKING, True, False), + # The deprecated FORWARD_ONLY category is a data change that is read like any other. + (True, SnapshotChangeCategory.FORWARD_ONLY, True, True), + (None, SnapshotChangeCategory.FORWARD_ONLY, True, False), ], ) def test_indirect_change_forward_only_with_stored_parent_history( make_snapshot, - virtual_environment_mode: VirtualEnvironmentMode, + earlier_version_flag: t.Optional[bool], + earlier_version_category: SnapshotChangeCategory, latest_version_forward_only: bool, expected_forward_only: bool, ): def parent_snapshot(value: int): - return make_snapshot( - SqlModel( - name="a", - query=parse_one(f"select {value} as col_a"), - virtual_environment_mode=virtual_environment_mode, - ) - ) + return make_snapshot(SqlModel(name="a", query=parse_one(f"select {value} as col_a"))) parent_baseline = parent_snapshot(0) parent_baseline.categorize_as(SnapshotChangeCategory.BREAKING) - stored_parent = parent_snapshot(1) - stored_parent.previous_versions = parent_baseline.all_versions + + # An earlier version of the stored parent, between the baseline and the latest version. + parent_earlier = parent_snapshot(1) + parent_earlier.previous_versions = parent_baseline.all_versions + parent_earlier.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) + parent_earlier.change_category = earlier_version_category + + stored_parent = parent_snapshot(2) + stored_parent.previous_versions = parent_earlier.all_versions stored_parent.categorize_as( SnapshotChangeCategory.BREAKING, forward_only=latest_version_forward_only ) @@ -4718,6 +4724,11 @@ def parent_snapshot(value: int): previous_gateway_managed_virtual_layer=False, gateway_managed_virtual_layer=False, environment_statements=[], + stored_parent_forward_only=( + {} + if earlier_version_flag is None + else {parent_earlier.snapshot_id: earlier_version_flag} + ), ) PlanBuilder(context_diff, forward_only=False).build()