Feat: Decouple forward-only from change categorization

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -239,8 +239,6 @@ def set_choice(self, snapshot: Snapshot, choice: SnapshotChangeCategory) -> Plan
             snapshot: The target snapshot.
             choice: The user decision on how to version the target snapshot and its children.
         """
-        if self._forward_only:
-            raise PlanError("Choice setting is not supported by a forward-only plan.")
         if not self._is_new_snapshot(snapshot):
             raise PlanError(
                 f"A choice can't be changed for the existing version of {snapshot.name}."
@@ -250,8 +248,6 @@ def set_choice(self, snapshot: Snapshot, choice: SnapshotChangeCategory) -> Plan
             and snapshot.snapshot_id not in self._context_diff.added
         ):
             raise PlanError(f"Only directly modified models can be categorized ({snapshot.name}).")
-        if snapshot.is_model and snapshot.model.forward_only:
-            raise PlanError(f"Forward-only model {snapshot.name} cannot be categorized manually.")
 
         self._choices[snapshot.snapshot_id] = choice
         self._latest_plan = None
@@ -369,8 +365,10 @@ def _build_restatements(
             restate_models = {
                 s.name
                 for s in self._context_diff.new_snapshots.values()
-                if s.is_materialized
-                and (self._forward_only or s.model.forward_only)
+                if s.is_model
+                and not s.is_symbolic
+                and (s.is_forward_only or s.model.forward_only)
+                and not s.is_no_preview
                 and (
                     # Metadata changes should not be previewed.
                     self._context_diff.directly_modified(s.name)
@@ -395,6 +393,9 @@ def _build_restatements(
         for s_id in dag:
             snapshot = self._context_diff.snapshots[s_id]
 
+            if is_preview and snapshot.is_no_preview:
+                continue
+
             # Since we are traversing the graph in topological order and the largest interval range is pushed down
             # the graph we just have to check our immediate parents in the graph and not the whole upstream graph.
             restating_parents = [
@@ -583,38 +584,35 @@ def _categorize_snapshots(
             if not snapshot or not self._is_new_snapshot(snapshot):
                 continue
 
+            forward_only = self._is_forward_only_change(s_id) or self._forward_only
+
             if s_id in self._choices:
-                snapshot.categorize_as(self._choices[s_id])
+                snapshot.categorize_as(self._choices[s_id], forward_only)
                 continue
 
             if s_id in self._context_diff.added:
-                snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
-            elif self._is_forward_only_change(s_id) or self._forward_only:
-                # In case of the forward only plan any modifications result in reuse of the
-                # previous version for non-seed models.
-                # New snapshots of seed models are considered non-breaking ones.
-                category = (
-                    SnapshotChangeCategory.NON_BREAKING
-                    if snapshot.is_seed
-                    else SnapshotChangeCategory.FORWARD_ONLY
-                )
-                # If the model kind changes mark as breaking
-                if snapshot.is_model and snapshot.name in self._context_diff.modified_snapshots:
-                    _, old = self._context_diff.modified_snapshots[snapshot.name]
-                    if _is_breaking_kind_change(old, snapshot):
-                        category = SnapshotChangeCategory.BREAKING
-
-                snapshot.categorize_as(category)
+                snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only)
             elif s_id.name in self._context_diff.modified_snapshots:
-                self._categorize_snapshot(snapshot, dag, indirectly_modified)
+                self._categorize_snapshot(snapshot, forward_only, dag, indirectly_modified)
 
     def _categorize_snapshot(
-        self, snapshot: Snapshot, dag: DAG[SnapshotId], indirectly_modified: SnapshotMapping
+        self,
+        snapshot: Snapshot,
+        forward_only: bool,
+        dag: DAG[SnapshotId],
+        indirectly_modified: SnapshotMapping,
     ) -> None:
         s_id = snapshot.snapshot_id
 
         if self._context_diff.directly_modified(s_id.name):
-            if self._auto_categorization_enabled:
+            new, old = self._context_diff.modified_snapshots[s_id.name]
+            if _is_breaking_kind_change(old, new):
+                snapshot.categorize_as(SnapshotChangeCategory.BREAKING, False)
+            elif self._auto_categorization_enabled:
+                if snapshot.is_seed:
+                    # Seed changes can't be forward-only.
+                    forward_only = False
+
                 s_id_with_missing_columns: t.Optional[SnapshotId] = None
                 this_sid_with_downstream = indirectly_modified.get(s_id, set()) | {s_id}
                 for downstream_s_id in this_sid_with_downstream:
@@ -626,18 +624,18 @@ def _categorize_snapshot(
                         s_id_with_missing_columns = downstream_s_id
                         break
 
-                new, old = self._context_diff.modified_snapshots[s_id.name]
                 if s_id_with_missing_columns is None:
                     change_category = categorize_change(new, old, config=self._categorizer_config)
                     if change_category is not None:
-                        snapshot.categorize_as(change_category)
+                        snapshot.categorize_as(change_category, forward_only)
                 else:
                     mode = self._categorizer_config.dict().get(
                         new.model.source_type, AutoCategorizationMode.OFF
                     )
                     if mode == AutoCategorizationMode.FULL:
-                        snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+                        snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only)
         elif self._context_diff.indirectly_modified(snapshot.name):
+            all_upstream_forward_only = set()
             all_upstream_categories = set()
             direct_parent_categories = set()
 
@@ -646,27 +644,30 @@ 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 p_id in snapshot.parents:
                         direct_parent_categories.add(parent.change_category)
 
-            if snapshot.is_model and snapshot.model.forward_only:
-                snapshot.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
-            elif direct_parent_categories.intersection(
+            if all_upstream_forward_only == {True} or (
+                snapshot.is_model and snapshot.model.forward_only
+            ):
+                forward_only = True
+
+            if direct_parent_categories.intersection(
                 {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
             ):
-                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
+                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING, forward_only)
             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)
+                snapshot.categorize_as(
+                    self._get_orphaned_indirect_change_category(snapshot), forward_only
+                )
             elif all_upstream_categories == {SnapshotChangeCategory.METADATA}:
-                snapshot.categorize_as(SnapshotChangeCategory.METADATA)
+                snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only)
             else:
-                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING)
+                snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING, forward_only)
         else:
             # Metadata updated.
-            snapshot.categorize_as(SnapshotChangeCategory.METADATA)
+            snapshot.categorize_as(SnapshotChangeCategory.METADATA, forward_only)
 
     def _get_orphaned_indirect_change_category(
         self, indirect_snapshot: Snapshot
@@ -769,10 +770,7 @@ def _is_forward_only_change(self, s_id: SnapshotId) -> bool:
             if snapshot.is_model and _is_breaking_kind_change(old, snapshot):
                 return False
         return (
-            snapshot.is_model
-            and snapshot.model.forward_only
-            and not snapshot.change_category
-            and bool(snapshot.previous_versions)
+            snapshot.is_model and snapshot.model.forward_only and bool(snapshot.previous_versions)
         )
 
     def _is_new_snapshot(self, snapshot: Snapshot) -> bool:
@@ -811,7 +809,7 @@ def _ensure_no_forward_only_revert(self) -> None:
                 and not candidate.model.forward_only
                 and promoted.is_forward_only
                 and not promoted.is_paused
-                and not candidate.reuses_previous_version
+                and not candidate.is_no_rebuild
                 and promoted.version == candidate.version
             ):
                 raise PlanError(

sqlmesh/core/snapshot/definition.py

@@ -76,6 +76,7 @@ class SnapshotChangeCategory(IntEnum):
 
     BREAKING = 1
     NON_BREAKING = 2
+    # FORWARD_ONLY category is deprecated and is kept for backwards compatibility.
     FORWARD_ONLY = 3
     INDIRECT_BREAKING = 4
     INDIRECT_NON_BREAKING = 5
@@ -336,6 +337,7 @@ class SnapshotInfoMixin(ModelKindMixin):
     base_table_name_override: t.Optional[str]
     dev_table_suffix: str
     table_naming_convention: TableNamingConvention = Field(default=TableNamingConvention.default)
+    forward_only: bool
 
     @cached_property
     def identifier(self) -> str:
@@ -383,7 +385,7 @@ def fully_qualified_table(self) -> t.Optional[exp.Table]:
 
     @property
     def is_forward_only(self) -> bool:
-        return self.change_category == SnapshotChangeCategory.FORWARD_ONLY
+        return self.forward_only or self.change_category == SnapshotChangeCategory.FORWARD_ONLY
 
     @property
     def is_metadata(self) -> bool:
@@ -394,9 +396,18 @@ def is_indirect_non_breaking(self) -> bool:
         return self.change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING
 
     @property
-    def reuses_previous_version(self) -> bool:
-        return self.change_category in (
-            SnapshotChangeCategory.FORWARD_ONLY,
+    def is_no_rebuild(self) -> bool:
+        """Returns true if this snapshot doesn't require a rebuild in production."""
+        return self.forward_only or self.change_category in (
+            SnapshotChangeCategory.FORWARD_ONLY,  # Backwards compatibility
+            SnapshotChangeCategory.METADATA,
+            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+        )
+
+    @property
+    def is_no_preview(self) -> bool:
+        """Returns true if this snapshot doesn't require a preview in development."""
+        return self.forward_only and self.change_category in (
             SnapshotChangeCategory.METADATA,
             SnapshotChangeCategory.INDIRECT_NON_BREAKING,
         )
@@ -487,6 +498,7 @@ class SnapshotTableInfo(PydanticModel, SnapshotInfoMixin, frozen=True):
     custom_materialization: t.Optional[str] = None
     dev_table_suffix: str
     model_gateway: t.Optional[str] = None
+    forward_only: bool = False
 
     def __lt__(self, other: SnapshotTableInfo) -> bool:
         return self.name < other.name
@@ -614,6 +626,7 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     table_naming_convention_: TableNamingConvention = Field(
         default=TableNamingConvention.default, alias="table_naming_convention"
     )
+    forward_only: bool = False
 
     @field_validator("ttl")
     @classmethod
@@ -1006,22 +1019,26 @@ def check_ready_intervals(
                 )
         return intervals
 
-    def categorize_as(self, category: SnapshotChangeCategory) -> None:
+    def categorize_as(self, category: SnapshotChangeCategory, forward_only: bool = False) -> None:
         """Assigns the given category to this snapshot.
 
         Args:
             category: The change category to assign to this snapshot.
+            forward_only: Whether or not this snapshot is applied going forward in production.
         """
+        assert category != SnapshotChangeCategory.FORWARD_ONLY, (
+            "FORWARD_ONLY change category is deprecated"
+        )
+
         self.dev_version_ = self.fingerprint.to_version()
-        reuse_previous_version = category in (
-            SnapshotChangeCategory.FORWARD_ONLY,
+        is_no_rebuild = forward_only or category in (
             SnapshotChangeCategory.INDIRECT_NON_BREAKING,
             SnapshotChangeCategory.METADATA,
         )
         if self.is_model and self.model.physical_version:
             # If the model has a pinned version then use that.
             self.version = self.model.physical_version
-        elif reuse_previous_version and self.previous_version:
+        elif is_no_rebuild and self.previous_version:
             previous_version = self.previous_version
             self.version = previous_version.data_version.version
             self.physical_schema_ = previous_version.physical_schema
@@ -1040,6 +1057,7 @@ def categorize_as(self, category: SnapshotChangeCategory) -> None:
             self.version = self.fingerprint.to_version()
 
         self.change_category = category
+        self.forward_only = forward_only
 
     @property
     def categorized(self) -> bool:
@@ -1220,6 +1238,7 @@ def table_info(self) -> SnapshotTableInfo:
             dev_table_suffix=self.dev_table_suffix,
             model_gateway=self.model_gateway,
             table_naming_convention=self.table_naming_convention,  # type: ignore
+            forward_only=self.forward_only,
         )
 
     @property

sqlmesh/core/snapshot/evaluator.py

@@ -338,7 +338,7 @@ def create(
                 continue
             deployability_flags = [True]
             if (
-                snapshot.reuses_previous_version
+                snapshot.is_no_rebuild
                 or snapshot.is_managed
                 or (snapshot.is_model and snapshot.model.forward_only)
                 or (deployability_index and not deployability_index.is_deployable(snapshot))

sqlmesh/core/state_sync/db/snapshot.py

@@ -733,14 +733,15 @@ class SharedVersionSnapshot(PydanticModel):
     disable_restatement: bool
     effective_from: t.Optional[TimeLike]
     raw_snapshot: t.Dict[str, t.Any]
+    forward_only: bool
 
     @property
     def snapshot_id(self) -> SnapshotId:
         return SnapshotId(name=self.name, identifier=self.identifier)
 
     @property
     def is_forward_only(self) -> bool:
-        return self.change_category == SnapshotChangeCategory.FORWARD_ONLY
+        return self.forward_only or self.change_category == SnapshotChangeCategory.FORWARD_ONLY
 
     @property
     def normalized_effective_from_ts(self) -> t.Optional[int]:
@@ -803,4 +804,5 @@ def from_snapshot_record(
             disable_restatement=raw_node.get("kind", {}).get("disable_restatement", False),
             effective_from=raw_snapshot.get("effective_from"),
             raw_snapshot=raw_snapshot,
+            forward_only=raw_snapshot.get("forward_only", False),
         )

tests/conftest.py

@@ -438,7 +438,7 @@ def _make_function(
                     metadata_hash="test_metadata_hash",
                 ),
                 version="test_version",
-                change_category=SnapshotChangeCategory.FORWARD_ONLY,
+                change_category=SnapshotChangeCategory.NON_BREAKING,
                 dev_table_suffix="dev",
             ),
         )

tests/core/analytics/test_collector.py

@@ -218,7 +218,7 @@ def test_on_snapshots_created(
         context.get_snapshot("sushi.waiter_revenue_by_day"),
         context.get_snapshot("sushi.top_waiters"),
     ]
-    new_snapshots[0].categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
+    new_snapshots[0].categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
     new_snapshots[0].effective_from = "2024-01-01"
     new_snapshots[0].version = "test_version"
 
@@ -239,7 +239,7 @@ def test_on_snapshots_created(
             "node_type": "model",
             "model_kind": "incremental_by_time_range",
             "is_sql": False,
-            "change_category": "forward_only",
+            "change_category": "breaking",
             "dialect": "duckdb",
             "audits_count": 0,
             "effective_from_set": True,

tests/core/engine_adapter/integration/test_integration.py

@@ -2087,7 +2087,9 @@ def _run_plan(sqlmesh_context: Context, environment: str = None) -> PlanResults:
     plan_1 = _run_plan(context)
 
     assert plan_1.snapshot_for(model_a).change_category == SnapshotChangeCategory.BREAKING
+    assert not plan_1.snapshot_for(model_a).is_forward_only
     assert plan_1.snapshot_for(model_b).change_category == SnapshotChangeCategory.BREAKING
+    assert not plan_1.snapshot_for(model_b).is_forward_only
 
     # so far so good, model_a should exist as a normal table, model b should be a managed table and the prod views should exist
     assert len(plan_1.schema_metadata.views) == 2
@@ -2134,8 +2136,10 @@ def _run_plan(sqlmesh_context: Context, environment: str = None) -> PlanResults:
 
     assert plan_2.plan.has_changes
     assert len(plan_2.plan.modified_snapshots) == 2
-    assert plan_2.snapshot_for(new_model_a).change_category == SnapshotChangeCategory.FORWARD_ONLY
+    assert plan_2.snapshot_for(new_model_a).change_category == SnapshotChangeCategory.NON_BREAKING
+    assert plan_2.snapshot_for(new_model_a).is_forward_only
     assert plan_2.snapshot_for(model_b).change_category == SnapshotChangeCategory.NON_BREAKING
+    assert not plan_2.snapshot_for(model_b).is_forward_only
 
     # verify that the new snapshots were created correctly
     # the forward-only change to model A should be in a new table separate from the one created in the first plan
@@ -2207,8 +2211,10 @@ def _run_plan(sqlmesh_context: Context, environment: str = None) -> PlanResults:
     plan_4 = _run_plan(context)
 
     assert plan_4.plan.has_changes
-    assert plan_4.snapshot_for(model_a).change_category == SnapshotChangeCategory.FORWARD_ONLY
+    assert plan_4.snapshot_for(model_a).change_category == SnapshotChangeCategory.NON_BREAKING
+    assert plan_4.snapshot_for(model_a).is_forward_only
     assert plan_4.snapshot_for(model_b).change_category == SnapshotChangeCategory.NON_BREAKING
+    assert not plan_4.snapshot_for(model_b).is_forward_only
 
     # verify the Model B table is created as a managed table in prod
     assert plan_4.table_name_for(model_b) == plan_3.table_name_for(