Fix: Correctly categorize an orphaned indirectly modified snapshot created as a result of a merge of 2 or more directly modified parents (#4329)

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -592,19 +592,97 @@ def _categorize_snapshot(
 
             if snapshot.is_model and snapshot.model.forward_only:
                 snapshot.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
-            elif not direct_parent_categories or direct_parent_categories.intersection(
+            elif direct_parent_categories.intersection(
                 {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
             ):
                 snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
+            elif not direct_parent_categories:
+                snapshot.categorize_as(self._get_orphaned_indirect_change_category(snapshot))
             elif SnapshotChangeCategory.FORWARD_ONLY in all_upstream_categories:
                 # FORWARD_ONLY must take precedence over INDIRECT_NON_BREAKING
                 snapshot.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
+            elif all_upstream_categories == {SnapshotChangeCategory.METADATA}:
+                snapshot.categorize_as(SnapshotChangeCategory.METADATA)
             else:
                 snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
         else:
             # Metadata updated.
             snapshot.categorize_as(SnapshotChangeCategory.METADATA)
 
+    def _get_orphaned_indirect_change_category(
+        self, indirect_snapshot: Snapshot
+    ) -> SnapshotChangeCategory:
+        """Sometimes an indirectly changed downstream snapshot ends up with no directly changed parents introduced in the same plan.
+        This may happen when 2 or more parent models were changed independently in different plans and then the changes were
+        merged together and applied in a single plan. As a result, a combination of 2 or more previously changed parents produces
+        a new downstream snapshot not previously seen.
+
+        This function is used to infer the correct change category for such downstream snapshots based on change categories of their parents.
+        """
+        previous_snapshot = self._context_diff.modified_snapshots[indirect_snapshot.name][1]
+        previous_parent_snapshot_ids = {p.name: p for p in previous_snapshot.parents}
+
+        current_parent_snapshots = [
+            self._context_diff.snapshots[p_id]
+            for p_id in indirect_snapshot.parents
+            if p_id in self._context_diff.snapshots
+        ]
+
+        indirect_category: t.Optional[SnapshotChangeCategory] = None
+        for current_parent_snapshot in current_parent_snapshots:
+            if current_parent_snapshot.name not in previous_parent_snapshot_ids:
+                # This is a new parent so falling back to INDIRECT_BREAKING
+                return SnapshotChangeCategory.INDIRECT_BREAKING
+            pevious_parent_snapshot_id = previous_parent_snapshot_ids[current_parent_snapshot.name]
+
+            if current_parent_snapshot.snapshot_id == pevious_parent_snapshot_id:
+                # There were no new versions of this parent since the previous version of this snapshot,
+                # 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:
+                # The previous parent is not in the historical chain so falling back to INDIRECT_BREAKING
+                return SnapshotChangeCategory.INDIRECT_BREAKING
+
+            if previous_parent_categories.intersection(
+                {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
+            ):
+                # One of the new parents in the chain was breaking so this indirect snapshot is breaking
+                return SnapshotChangeCategory.INDIRECT_BREAKING
+
+            if SnapshotChangeCategory.FORWARD_ONLY in previous_parent_categories:
+                # One of the new parents in the chain was forward-only so this indirect snapshot is forward-only
+                indirect_category = SnapshotChangeCategory.FORWARD_ONLY
+            elif (
+                previous_parent_categories.intersection(
+                    {
+                        SnapshotChangeCategory.NON_BREAKING,
+                        SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+                    }
+                )
+                and indirect_category != SnapshotChangeCategory.FORWARD_ONLY
+            ):
+                # All changes in the chain were non-breaking so this indirect snapshot can be non-breaking too
+                indirect_category = SnapshotChangeCategory.INDIRECT_NON_BREAKING
+            elif (
+                previous_parent_categories == {SnapshotChangeCategory.METADATA}
+                and indirect_category is None
+            ):
+                # All changes in the chain were metadata so this indirect snapshot can be metadata too
+                indirect_category = SnapshotChangeCategory.METADATA
+
+        return indirect_category or SnapshotChangeCategory.INDIRECT_BREAKING
+
     def _apply_effective_from(self) -> None:
         if self._effective_from:
             if not self._forward_only:

tests/core/test_integration.py

@@ -2247,6 +2247,114 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot_migration(
     assert count > 0
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+@pytest.mark.parametrize(
+    "parent_a_category,parent_b_category,expected_child_category",
+    [
+        (
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.INDIRECT_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.INDIRECT_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.METADATA,
+        ),
+        (
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.METADATA,
+        ),
+        (
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.INDIRECT_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.NON_BREAKING,
+            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.METADATA,
+        ),
+        (
+            SnapshotChangeCategory.FORWARD_ONLY,
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.INDIRECT_BREAKING,
+        ),
+        (
+            SnapshotChangeCategory.BREAKING,
+            SnapshotChangeCategory.FORWARD_ONLY,
+            SnapshotChangeCategory.FORWARD_ONLY,
+        ),
+        (
+            SnapshotChangeCategory.FORWARD_ONLY,
+            SnapshotChangeCategory.FORWARD_ONLY,
+            SnapshotChangeCategory.FORWARD_ONLY,
+        ),
+    ],
+)
+def test_rebase_two_changed_parents(
+    init_and_plan_context: t.Callable,
+    parent_a_category: SnapshotChangeCategory,  # This change is deployed to prod first
+    parent_b_category: SnapshotChangeCategory,  # This change is deployed to prod second
+    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")
+
+    # Make change A and deploy it to dev_a
+    context.upsert_model(initial_model_a.name, stamp="1")
+    plan_builder = context.plan_builder("dev_a", skip_tests=True)
+    plan_builder.set_choice(context.get_snapshot(initial_model_a.name), parent_a_category)
+    context.apply(plan_builder.build())
+
+    # Make change B and deploy it to dev_b
+    context.upsert_model(initial_model_a)
+    context.upsert_model(initial_model_b.name, stamp="1")
+    plan_builder = context.plan_builder("dev_b", skip_tests=True)
+    plan_builder.set_choice(context.get_snapshot(initial_model_b.name), parent_b_category)
+    context.apply(plan_builder.build())
+
+    # Deploy change A to prod
+    context.upsert_model(initial_model_a.name, stamp="1")
+    context.upsert_model(initial_model_b)
+    context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True)
+
+    # Apply change B in addition to A and plan against prod
+    context.upsert_model(initial_model_b.name, stamp="1")
+    plan = context.plan_builder("prod", skip_tests=True).build()
+
+    # Validate the category of child snapshots
+    direct_child_snapshot = plan.snapshots[context.get_snapshot("sushi.order_items").snapshot_id]
+    assert direct_child_snapshot.change_category == expected_child_category
+
+    indirect_child_snapshot = plan.snapshots[context.get_snapshot("sushi.top_waiters").snapshot_id]
+    assert indirect_child_snapshot.change_category == expected_child_category
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_unaligned_start_snapshot_with_non_deployable_downstream(init_and_plan_context: t.Callable):
     context, _ = init_and_plan_context("examples/sushi")