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 a6307a9ffd..a13362c4c3 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 ( @@ -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,15 +744,33 @@ 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) + # 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) + ) + + 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 ): forward_only = True - if direct_parent_categories.intersection( + if (direct_parent_categories | stored_parent_categories).intersection( {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING} ): snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING, forward_only) @@ -751,7 +778,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 +788,112 @@ def _categorize_snapshot( # Metadata updated. snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only) + def _get_parent_versions_since( + self, parent: Snapshot, previous_parent_id: SnapshotId + ) -> 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. + """ + versions: t.List[SnapshotDataVersion] = [] + for pv in reversed(parent.all_versions): + if pv.snapshot_id(parent.name) == previous_parent_id: + 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 _get_known_category(version: SnapshotDataVersion) -> SnapshotChangeCategory: + """Returns the category of the version, or BREAKING if it isn't one that is assigned today. + + 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. + """ + 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 + ) -> 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 + 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() + 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) + 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_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) + continue + + 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, category in version_categories + if category + in (SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING) + ] + if 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( self, indirect_snapshot: Snapshot ) -> SnapshotChangeCategory: @@ -790,17 +925,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..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, @@ -1515,3 +1516,301 @@ 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") +@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) + + 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, 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 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") +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) + + +@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) + + +@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 add85d2eec..8428e9130c 100644 --- a/tests/core/test_plan.py +++ b/tests/core/test_plan.py @@ -4510,3 +4510,228 @@ 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 + + +@pytest.mark.parametrize( + "earlier_version_flag, earlier_version_category, latest_version_forward_only, expected_forward_only", + [ + # 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, + 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"))) + + parent_baseline = parent_snapshot(0) + parent_baseline.categorize_as(SnapshotChangeCategory.BREAKING) + + # 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 + ) + + 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=[], + 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() + + assert child_new.change_category == SnapshotChangeCategory.INDIRECT_BREAKING + assert child_new.is_forward_only == expected_forward_only