Fix!: Avoid using rendered query when computing the data hash

Author: izeigerman

sqlmesh/core/context_diff.py

@@ -435,7 +435,7 @@ def directly_modified(self, name: str) -> bool:
             return False
 
         current, previous = self.modified_snapshots[name]
-        return current.fingerprint.data_hash != previous.fingerprint.data_hash
+        return current.is_directly_modified(previous)
 
     def indirectly_modified(self, name: str) -> bool:
         """Returns whether or not a node was indirectly modified in this context.
@@ -451,10 +451,7 @@ def indirectly_modified(self, name: str) -> bool:
             return False
 
         current, previous = self.modified_snapshots[name]
-        return (
-            current.fingerprint.data_hash == previous.fingerprint.data_hash
-            and current.fingerprint.parent_data_hash != previous.fingerprint.parent_data_hash
-        )
+        return current.is_indirectly_modified(previous)
 
     def metadata_updated(self, name: str) -> bool:
         """Returns whether or not the given node's metadata has been updated.
@@ -470,7 +467,7 @@ def metadata_updated(self, name: str) -> bool:
             return False
 
         current, previous = self.modified_snapshots[name]
-        return current.fingerprint.metadata_hash != previous.fingerprint.metadata_hash
+        return current.is_metadata_updated(previous)
 
     def text_diff(self, name: str) -> str:
         """Finds the difference of a node between the current and remote environment.

sqlmesh/core/model/definition.py

@@ -62,6 +62,7 @@
 
 if t.TYPE_CHECKING:
     from sqlglot.dialects.dialect import DialectType
+    from sqlmesh.core.node import _Node
     from sqlmesh.core._typing import Self, TableName, SessionProperties
     from sqlmesh.core.context import ExecutionContext
     from sqlmesh.core.engine_adapter import EngineAdapter
@@ -1278,6 +1279,7 @@ class SqlModel(_Model):
     source_type: t.Literal["sql"] = "sql"
 
     _columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
+    _is_metadata_only_change_cache: t.Dict[int, bool] = {}
 
     def __getstate__(self) -> t.Dict[t.Any, t.Any]:
         state = super().__getstate__()
@@ -1500,6 +1502,27 @@ def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
 
         return False
 
+    def is_metadata_only_change(self, previous: _Node) -> bool:
+        if self._is_metadata_only_change_cache.get(id(previous), None) is not None:
+            return self._is_metadata_only_change_cache[id(previous)]
+
+        if (
+            not isinstance(previous, SqlModel)
+            or self.metadata_hash == previous.metadata_hash
+            or self._data_hash_values_no_query != previous._data_hash_values_no_query
+        ):
+            is_metadata_change = False
+        else:
+            # If the rendered queries are the same, then this is a metadata only change
+            this_rendered_query = self.render_query()
+            previous_rendered_query = previous.render_query()
+            is_metadata_change = (
+                this_rendered_query is not None and this_rendered_query == previous_rendered_query
+            )
+
+        self._is_metadata_only_change_cache[id(previous)] = is_metadata_change
+        return is_metadata_change
+
     @cached_property
     def _query_renderer(self) -> QueryRenderer:
         no_quote_identifiers = self.kind.is_view and self.dialect in ("trino", "spark")
@@ -1519,17 +1542,22 @@ def _query_renderer(self) -> QueryRenderer:
         )
 
     @property
-    def _data_hash_values(self) -> t.List[str]:
-        data = super()._data_hash_values
+    def _data_hash_values_no_query(self) -> t.List[str]:
+        return [
+            *super()._data_hash_values,
+            *self.jinja_macros.data_hash_values,
+        ]
 
-        query = self.render_query() or self.query
-        data.append(gen(query))
-        data.extend(self.jinja_macros.data_hash_values)
-        return data
+    @property
+    def _data_hash_values(self) -> t.List[str]:
+        return [
+            *self._data_hash_values_no_query,
+            gen(self.query, comments=False),
+        ]
 
     @property
     def _additional_metadata(self) -> t.List[str]:
-        return [*super()._additional_metadata, gen(self.query)]
+        return [*super()._additional_metadata, gen(self.query, comments=True)]
 
     @property
     def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]:

sqlmesh/core/node.py

@@ -307,16 +307,6 @@ def batch_concurrency(self) -> t.Optional[int]:
         """The maximal number of batches that can run concurrently for a backfill."""
         return None
 
-    @property
-    def data_hash(self) -> str:
-        """
-        Computes the data hash for the node.
-
-        Returns:
-            The data hash for the node.
-        """
-        raise NotImplementedError
-
     @property
     def interval_unit(self) -> IntervalUnit:
         """Returns the interval unit using which data intervals are computed for this node."""
@@ -332,6 +322,16 @@ def depends_on(self) -> t.Set[str]:
     def fqn(self) -> str:
         return self.name
 
+    @property
+    def data_hash(self) -> str:
+        """
+        Computes the data hash for the node.
+
+        Returns:
+            The data hash for the node.
+        """
+        raise NotImplementedError
+
     @property
     def metadata_hash(self) -> str:
         """
@@ -342,6 +342,30 @@ def metadata_hash(self) -> str:
         """
         raise NotImplementedError
 
+    def is_metadata_only_change(self, previous: _Node) -> bool:
+        """Determines if this node is a metadata only change in relation to the `previous` node.
+
+        Args:
+            previous: The previous node to compare against.
+
+        Returns:
+            True if this node is a metadata only change, False otherwise.
+        """
+        return self.data_hash == previous.data_hash and self.metadata_hash != previous.metadata_hash
+
+    def is_data_change(self, previous: _Node) -> bool:
+        """Determines if this node is a data change in relation to the `previous` node.
+
+        Args:
+            previous: The previous node to compare against.
+
+        Returns:
+            True if this node is a data change, False otherwise.
+        """
+        return (
+            self.data_hash != previous.data_hash or self.metadata_hash != previous.metadata_hash
+        ) and not self.is_metadata_only_change(previous)
+
     def croniter(self, value: TimeLike) -> CroniterCache:
         if self._croniter is None:
             self._croniter = CroniterCache(self.cron, value, tz=self.cron_tz)

sqlmesh/core/snapshot/categorizer.py

@@ -47,11 +47,12 @@ def categorize_change(
     if type(new_model) != type(old_model):
         return default_category
 
-    if new.fingerprint.data_hash == old.fingerprint.data_hash:
-        if new.fingerprint.metadata_hash == old.fingerprint.metadata_hash:
-            raise SQLMeshError(
-                f"{new} is unmodified or indirectly modified and should not be categorized"
-            )
+    if new.fingerprint == old.fingerprint:
+        raise SQLMeshError(
+            f"{new} is unmodified or indirectly modified and should not be categorized"
+        )
+
+    if not new.is_directly_modified(old):
         if new.fingerprint.parent_data_hash == old.fingerprint.parent_data_hash:
             return SnapshotChangeCategory.NON_BREAKING
         return None

sqlmesh/core/snapshot/definition.py

@@ -1230,6 +1230,21 @@ def apply_pending_restatement_intervals(self) -> None:
             )
             self.intervals = remove_interval(self.intervals, *pending_restatement_interval)
 
+    def is_directly_modified(self, other: Snapshot) -> bool:
+        """Returns whether or not this snapshot is directly modified in relation to the other snapshot."""
+        return self.node.is_data_change(other.node)
+
+    def is_indirectly_modified(self, other: Snapshot) -> bool:
+        """Returns whether or not this snapshot is indirectly modified in relation to the other snapshot."""
+        return (
+            self.fingerprint.parent_data_hash != other.fingerprint.parent_data_hash
+            and not self.node.is_data_change(other.node)
+        )
+
+    def is_metadata_updated(self, other: Snapshot) -> bool:
+        """Returns whether or not this snapshot contains metadata changes in relation to the other snapshot."""
+        return self.fingerprint.metadata_hash != other.fingerprint.metadata_hash
+
     @property
     def physical_schema(self) -> str:
         if self.physical_schema_ is not None:

sqlmesh/migrations/v0092_use_unrendered_query_in_fingerprint.py

@@ -0,0 +1,5 @@
+"""Use the unrendered query when computing the model fingerprint."""
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    pass

tests/core/test_integration.py

@@ -4681,12 +4681,12 @@ def test_plan_repairs_unrenderable_snapshot_state(
         f"name = '{target_snapshot.name}' AND identifier = '{target_snapshot.identifier}'",
     )
 
+    context.clear_caches()
+    target_snapshot_in_state = context.state_sync.get_snapshots([target_snapshot.snapshot_id])[
+        target_snapshot.snapshot_id
+    ]
+
     with pytest.raises(Exception):
-        context_copy = context.copy()
-        context_copy.clear_caches()
-        target_snapshot_in_state = context_copy.state_sync.get_snapshots(
-            [target_snapshot.snapshot_id]
-        )[target_snapshot.snapshot_id]
         target_snapshot_in_state.model.render_query_or_raise()
 
     # Repair the snapshot by creating a new version of it
@@ -4695,11 +4695,11 @@ def test_plan_repairs_unrenderable_snapshot_state(
 
     plan_builder = context.plan_builder("prod", forward_only=forward_only)
     plan = plan_builder.build()
-    assert plan.directly_modified == {target_snapshot.snapshot_id}
     if not forward_only:
         assert target_snapshot.snapshot_id in {i.snapshot_id for i in plan.missing_intervals}
-        plan_builder.set_choice(target_snapshot, SnapshotChangeCategory.NON_BREAKING)
-        plan = plan_builder.build()
+    assert plan.directly_modified == {target_snapshot.snapshot_id}
+    plan_builder.set_choice(target_snapshot, SnapshotChangeCategory.NON_BREAKING)
+    plan = plan_builder.build()
 
     context.apply(plan)
 

tests/core/test_model.py

@@ -5735,7 +5735,7 @@ def test_default_catalog_sql(assert_exp_eq):
     The system is not designed to actually support having an engine that doesn't support default catalog
     to start supporting it or the reverse of that. If that did happen then bugs would occur.
     """
-    HASH_WITH_CATALOG = "1269513823"
+    HASH_WITH_CATALOG = "3443912775"
 
     # Test setting default catalog doesn't change hash if it matches existing logic
     expressions = d.parse(

tests/core/test_selector.py

@@ -301,7 +301,7 @@ def test_select_change_schema(mocker: MockerFixture, make_snapshot):
     selector = Selector(state_reader_mock, local_models)
 
     selected = selector.select_models(["db.parent"], env_name)
-    assert selected[local_child.fqn].data_hash != child.data_hash
+    assert selected[local_child.fqn].render_query() != child.render_query()
 
     _assert_models_equal(
         selected,

tests/core/test_snapshot.py

@@ -913,7 +913,7 @@ def test_fingerprint(model: Model, parent_model: Model):
     fingerprint = fingerprint_from_node(model, nodes={})
 
     original_fingerprint = SnapshotFingerprint(
-        data_hash="3301649319",
+        data_hash="1698409777",
         metadata_hash="3575333731",
     )
 
@@ -1013,7 +1013,7 @@ def test_fingerprint_jinja_macros(model: Model):
         }
     )
     original_fingerprint = SnapshotFingerprint(
-        data_hash="2908339239",
+        data_hash="343517722",
         metadata_hash="3575333731",
     )