From b4987b4a92d57575627ea7dde7c23b03d624410b Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 10:42:45 -0700
Subject: [PATCH 01/11] Add semantic node fingerprints

Derive node digests and canonical comparisons from existing change tiers and equality rules so semantic classification remains server-owned.
---
 .../datajunction_server/models/deployment.py  | 484 ++++++++++++++----
 .../tests/models/deployment_test.py           | 443 ++++++++++++++++
 2 files changed, 839 insertions(+), 88 deletions(-)

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index 2bf7081e5..745bb7f2a 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -1,7 +1,9 @@
+import hashlib
 import json
+import math
 from collections.abc import Iterable
 from enum import Enum, IntEnum
-from typing import Annotated, Any, ClassVar, Literal
+from typing import Annotated, Any, ClassVar, Literal, get_args, get_origin
 
 from pydantic import (
     AliasChoices,
@@ -68,6 +70,19 @@ class ChangeTier(IntEnum):
     MAJOR = 20
 
 
+class SemanticFingerprint(BaseModel):
+    """A digest of a node's semantic definition."""
+
+    digest: str = Field(
+        min_length=64,
+        max_length=64,
+        pattern=r"^[0-9a-f]+$",
+    )
+
+
+_SEMANTIC_FINGERPRINT_VERSION = 1
+
+
 def fold_change_tiers(tiers: Iterable[ChangeTier]) -> ChangeTier:
     """
     Reduce the tiers of several individual changes to the tier of the change as a
@@ -473,10 +488,13 @@ class DimensionLinkSpec(BaseModel):
     role: str | None = None
     namespace: str | None = Field(default=None, exclude=True)
 
+    def _comparison_key(self) -> tuple[Any, ...]:
+        return (self.type, self.role)
+
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, DimensionLinkSpec):
             return False  # pragma: no cover
-        return self.type == other.type and self.role == other.role
+        return self._comparison_key() == other._comparison_key()
 
 
 class DimensionJoinLinkSpec(DimensionLinkSpec):
@@ -517,31 +535,23 @@ def rendered_join_on(self) -> str | None:
         )
 
     def __hash__(self) -> int:
-        return hash(
-            (
-                self.type,
-                self.role,
-                self.rendered_dimension_node,
-                self.join_type,
-                self.join_cardinality,
-                self.rendered_join_on,
-                self.node_column,
-                self.default_value,
-            ),
+        return hash(self._comparison_key())
+
+    def _comparison_key(self) -> tuple[Any, ...]:
+        return (
+            *super()._comparison_key(),
+            self.rendered_dimension_node,
+            self.join_type,
+            self.join_cardinality,
+            self.rendered_join_on,
+            self.node_column,
+            self.default_value,
         )
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, DimensionJoinLinkSpec):
             return False  # pragma: no cover
-        return (
-            super().__eq__(other)
-            and self.rendered_dimension_node == other.rendered_dimension_node
-            and self.join_type == other.join_type
-            and self.join_cardinality == other.join_cardinality
-            and self.rendered_join_on == other.rendered_join_on
-            and self.node_column == other.node_column
-            and self.default_value == other.default_value
-        )
+        return self._comparison_key() == other._comparison_key()
 
 
 class DimensionReferenceLinkSpec(DimensionLinkSpec):
@@ -570,25 +580,20 @@ def dimension_attribute(self) -> str:
         return self.dimension.rsplit(".", 1)[-1]
 
     def __hash__(self) -> int:
-        return hash(
-            (
-                self.type,
-                self.role,
-                self.rendered_dimension_node,
-                self.dimension_attribute,
-                self.node_column,
-            ),
+        return hash(self._comparison_key())
+
+    def _comparison_key(self) -> tuple[Any, ...]:
+        return (
+            *super()._comparison_key(),
+            self.rendered_dimension_node,
+            self.dimension_attribute,
+            self.node_column,
         )
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, DimensionReferenceLinkSpec):
             return False
-        return (
-            super().__eq__(other)
-            and self.rendered_dimension_node == other.rendered_dimension_node
-            and self.dimension_attribute == other.dimension_attribute
-            and self.node_column == other.node_column
-        )
+        return self._comparison_key() == other._comparison_key()
 
 
 def render_prefixes(parameterized_string: str, prefix: str | None = None) -> str:
@@ -716,6 +721,37 @@ def rendered_spec(self) -> "NodeSpec":
         rendered_json = json.dumps(raw).replace("${prefix}", prefix)
         return self.__class__.model_validate_json(rendered_json)
 
+    def semantic_fingerprint(
+        self,
+        version: int = _SEMANTIC_FINGERPRINT_VERSION,
+        *,
+        resolved_columns: list[ColumnSpec] | None = None,
+    ) -> SemanticFingerprint:
+        """Return a fingerprint of this node's semantic definition."""
+        if type(version) is not int or version != _SEMANTIC_FINGERPRINT_VERSION:
+            raise ValueError(f"Unsupported semantic fingerprint version: {version}")
+        payload = _semantic_fingerprint_payload(
+            self,
+            resolved_columns=resolved_columns,
+        )
+        digest = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
+        return SemanticFingerprint(digest=digest)
+
+    def canonical_diff(
+        self,
+        other: "NodeSpec",
+        *,
+        resolved_columns: list[ColumnSpec] | None = None,
+        other_resolved_columns: list[ColumnSpec] | None = None,
+    ) -> tuple[list[str], list[str]]:
+        """Compare two specs using the same canonical values as fingerprints."""
+        return _canonical_node_diff(
+            self,
+            other,
+            resolved_columns=resolved_columns,
+            other_resolved_columns=other_resolved_columns,
+        )
+
     def diff(self, other: "NodeSpec") -> list[str]:
         """
         Return a list of fields that differ between this and another NodeSpec.
@@ -864,21 +900,21 @@ def links_mapping(self) -> dict[tuple[str, str | None], DimensionLinkSpec]:
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, LinkableNodeSpec):
             return False  # pragma: no cover
-        dimension_links_equal = sorted(
-            self.dimension_links or [],
-            key=lambda link: (link.rendered_dimension_node, link.role or ""),
-        ) == sorted(
-            other.dimension_links or [],
-            key=lambda link: (link.rendered_dimension_node, link.role or ""),
-        )
         return (
             super().__eq__(other)
             and eq_columns(
                 self.columns,
                 other.columns,
-                compare_types=True if self.node_type == NodeType.SOURCE else False,
+                compare_types=self.node_type == NodeType.SOURCE,
+            )
+            and _canonical_dimension_links(
+                self.dimension_links,
+                preserve_order=False,
+            )
+            == _canonical_dimension_links(
+                other.dimension_links,
+                preserve_order=False,
             )
-            and dimension_links_equal
             and set(self.primary_key or []) == set(other.primary_key or [])
         )
 
@@ -1298,15 +1334,9 @@ def __eq__(self, other: object) -> bool:
 
         # Compare only partition config for user-specified columns.
         # Cube element columns (types, order, attributes) are auto-derived and ignored.
-        incoming_partitions = {
-            col.name: col.partition for col in self.rendered_columns if col.partition
-        }
-        existing_partitions = {
-            col.name: col.partition
-            for col in (other.rendered_columns or [])
-            if col.partition
-        }
-        return incoming_partitions == existing_partitions
+        return _canonical_cube_columns(
+            self.rendered_columns,
+        ) == _canonical_cube_columns(other.rendered_columns)
 
 
 NodeUnion = Annotated[
@@ -1315,6 +1345,304 @@ def __eq__(self, other: object) -> bool:
 ]
 
 
+def _canonical_json(value: Any) -> str:
+    """Serialize a fingerprint value deterministically."""
+    return json.dumps(
+        value,
+        allow_nan=False,
+        ensure_ascii=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
+def _canonical_value(value: Any) -> Any:
+    """Convert supported values to deterministic JSON-compatible values."""
+    if isinstance(value, Enum):
+        return _canonical_value(value.value)
+    if isinstance(value, BaseModel):
+        return _canonical_value(value.model_dump(mode="python"))
+    if isinstance(value, dict):
+        if any(not isinstance(key, str) for key in value):
+            raise TypeError("Semantic fingerprint mappings require string keys")
+        return {key: _canonical_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_canonical_value(item) for item in value]
+    if isinstance(value, float) and not math.isfinite(value):
+        raise ValueError("Semantic fingerprint values must be finite")
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    raise TypeError(
+        f"Unsupported semantic fingerprint value: {type(value).__name__}",
+    )
+
+
+def _canonical_sql_ast(query_ast: Any) -> Any:
+    """Serialize parsed SQL from AST fields without invoking SQL rendering."""
+    from datajunction_server.sql.parsing.ast import Node as SQLNode
+    from datajunction_server.sql.parsing.types import ColumnType
+
+    def serialize(value: Any) -> Any:
+        if isinstance(value, SQLNode):
+            return {
+                "type": f"{type(value).__module__}.{type(value).__qualname__}",
+                "fields": {
+                    name: serialize(field_value)
+                    for name, field_value in value.fields(
+                        flat=False,
+                        nodes_only=False,
+                        obfuscated=False,
+                        nones=True,
+                        named=True,
+                    )
+                },
+            }
+        if isinstance(value, ColumnType):
+            return {
+                "type": f"{type(value).__module__}.{type(value).__qualname__}",
+                "value": str(value),
+            }
+        if isinstance(value, Enum):
+            return {
+                "type": f"{type(value).__module__}.{type(value).__qualname__}",
+                "value": serialize(value.value),
+            }
+        if isinstance(value, (list, tuple)):
+            return [serialize(item) for item in value]
+        return _canonical_value(value)
+
+    return serialize(query_ast)
+
+
+def _canonical_sequence(
+    values: Iterable[Any],
+    *,
+    preserve_order: bool,
+) -> list[Any]:
+    """Canonicalize a sequence while removing duplicate semantic values."""
+    unique: dict[str, Any] = {}
+    for value in values:
+        canonical = _canonical_value(value)
+        unique.setdefault(_canonical_json(canonical), canonical)
+    return (
+        list(unique.values())
+        if preserve_order
+        else [unique[key] for key in sorted(unique)]
+    )
+
+
+def _annotation_contains_list(annotation: Any) -> bool:
+    return get_origin(annotation) is list or any(
+        _annotation_contains_list(argument) for argument in get_args(annotation)
+    )
+
+
+def _normalized_column(
+    column: ColumnSpec | None,
+    name: str,
+    fallback_type: str | None,
+    compare_types: bool,
+) -> ColumnSpec:
+    normalized = (
+        column.model_copy()
+        if column
+        else ColumnSpec(
+            name=name,
+            display_name=labelize(name),
+            type=fallback_type or "",
+            attributes=[],
+        )
+    )
+    normalized.display_name = normalized.display_name or labelize(name)
+    normalized.description = normalized.description or ""
+    normalized.attributes = sorted(set(normalized.attributes) - {"primary_key"})
+    if not compare_types:
+        normalized.type = ""
+    return normalized
+
+
+def _canonical_columns(
+    columns: list[ColumnSpec] | None,
+    *,
+    compare_types: bool,
+) -> list[Any]:
+    column_map = {column.name: column for column in columns or []}
+    canonical = []
+    for name in sorted(column_map):
+        column = column_map[name]
+        normalized = _normalized_column(column, name, column.type, compare_types)
+        if not compare_types:
+            default = _normalized_column(None, name, column.type, compare_types)
+            if normalized == default:
+                continue
+        canonical.append(_canonical_value(normalized))
+    return canonical
+
+
+def _canonical_dimension_links(
+    links: list[DimensionJoinLinkSpec | DimensionReferenceLinkSpec] | None,
+    *,
+    preserve_order: bool,
+) -> list[Any]:
+    return _canonical_sequence(
+        (link._comparison_key() for link in links or []),
+        preserve_order=preserve_order,
+    )
+
+
+def _canonical_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]:
+    return {
+        column.name: _canonical_value(column.partition)
+        for column in columns or []
+        if column.partition
+    }
+
+
+def _canonical_field_value(
+    spec: NodeSpec,
+    field: str,
+    *,
+    resolved_columns: list[ColumnSpec] | None = None,
+    preserve_order: bool = False,
+) -> Any:
+    value = getattr(spec, field)
+    if field == "query":
+        return (
+            _canonical_sql_ast(spec.query_ast)
+            if spec.query_ast is not None
+            else spec.rendered_query
+        )
+    if field == "columns":
+        if isinstance(spec, CubeSpec):
+            return _canonical_cube_columns(spec.rendered_columns)
+        return _canonical_columns(
+            resolved_columns
+            if isinstance(spec, SourceSpec) and resolved_columns is not None
+            else value,
+            compare_types=isinstance(spec, SourceSpec),
+        )
+    if field == "dimension_links" and isinstance(spec, LinkableNodeSpec):
+        return _canonical_dimension_links(
+            spec.dimension_links,
+            preserve_order=preserve_order,
+        )
+    if field == "unit_enum" and isinstance(spec, MetricSpec):
+        return _canonical_value(spec._canonical_unit())
+    if field == "direction" and isinstance(spec, MetricSpec):
+        value = value or MetricDirection.NEUTRAL
+    if field == "description":
+        value = value or None
+    if field == "custom_metadata":
+        value = value or {}
+    if value is None and _annotation_contains_list(
+        type(spec).model_fields[field].annotation,
+    ):
+        value = []
+
+    canonical = _canonical_value(value)
+    return (
+        _canonical_sequence(canonical, preserve_order=preserve_order)
+        if isinstance(canonical, list)
+        else canonical
+    )
+
+
+def _semantic_fingerprint_payload(
+    spec: NodeSpec,
+    *,
+    resolved_columns: list[ColumnSpec] | None,
+) -> dict[str, Any]:
+    for field, field_info in type(spec).model_fields.items():
+        if field in {"name", "namespace", "node_type"} or field_info.exclude is True:
+            continue
+        if type(spec).field_change_tier(field) == ChangeTier.MAJOR:
+            _canonical_json(_canonical_value(getattr(spec, field)))
+
+    rendered = spec.rendered_spec()
+    fields = {}
+    for field, field_info in type(rendered).model_fields.items():
+        if field in {"name", "namespace", "node_type"}:
+            continue
+        if field_info.exclude is True:
+            continue
+        if type(rendered).field_change_tier(field) != ChangeTier.MAJOR:
+            continue
+        fields[field] = _canonical_field_value(
+            rendered,
+            field,
+            resolved_columns=resolved_columns,
+            preserve_order=(
+                type(rendered).field_order_change_tier(field) == ChangeTier.MAJOR
+            ),
+        )
+    return {
+        "domain": "datajunction/node-semantic",
+        "node_type": _canonical_value(rendered.node_type),
+        "fields": fields,
+    }
+
+
+def _canonical_node_diff(
+    one: NodeSpec,
+    two: NodeSpec,
+    *,
+    resolved_columns: list[ColumnSpec] | None,
+    other_resolved_columns: list[ColumnSpec] | None,
+) -> tuple[list[str], list[str]]:
+    if one.node_type != two.node_type:
+        return ["node_type"], []
+
+    rendered_one = one.rendered_spec()
+    rendered_two = two.rendered_spec()
+
+    changed_fields = []
+    reordered_fields = []
+    for field, field_info in type(rendered_two).model_fields.items():
+        if field in {"name", "namespace", "node_type"}:
+            continue
+        if isinstance(rendered_two, MetricSpec) and field == "unit_structured":
+            continue
+        if field_info.exclude is True and field != "unit_enum":
+            continue
+        if type(rendered_two).field_change_tier(field) == ChangeTier.NONE:
+            continue
+        if field == "display_name" and getattr(rendered_two, field) is None:
+            continue
+
+        left = _canonical_field_value(
+            rendered_one,
+            field,
+            resolved_columns=resolved_columns,
+        )
+        right = _canonical_field_value(
+            rendered_two,
+            field,
+            resolved_columns=other_resolved_columns,
+        )
+        if left != right:
+            changed_fields.append(field)
+            continue
+
+        if type(rendered_two).field_order_change_tier(field) == ChangeTier.NONE:
+            continue
+        left_ordered = _canonical_field_value(
+            rendered_one,
+            field,
+            resolved_columns=resolved_columns,
+            preserve_order=True,
+        )
+        right_ordered = _canonical_field_value(
+            rendered_two,
+            field,
+            resolved_columns=other_resolved_columns,
+            preserve_order=True,
+        )
+        if left_ordered != right_ordered:
+            reordered_fields.append(field)
+
+    return changed_fields, reordered_fields
+
+
 def _norm(v: Any) -> Any:
     """Normalize falsy string/None to None so that '' and None compare equal."""
     if isinstance(v, str):
@@ -1366,7 +1694,7 @@ def diff(
     """
     return [
         field
-        for field in one.model_fields.keys()
+        for field in one.model_fields
         if field not in (ignore_fields or [])
         and hasattr(one, field)
         and hasattr(two, field)
@@ -1675,39 +2003,19 @@ def eq_columns(
     if compare_types and a and b and set(a_map.keys()) != set(b_map.keys()):
         return False
     a_cols, b_cols = [], []
-    for col_name in set(a_map.keys()).union(set(b_map.keys())):
-        a_col = a_map.get(col_name).model_copy() if a_map.get(col_name) else None  # type: ignore
-        b_col = b_map.get(col_name).model_copy() if b_map.get(col_name) else None  # type: ignore
-        if not a_col:
-            a_col = ColumnSpec(
-                name=col_name,
-                display_name=labelize(col_name),
-                type=b_col.type if b_col else "",
-                attributes=[],
-            )
-        if not a_col.display_name:
-            a_col.display_name = labelize(col_name)
-        if not a_col.description:
-            a_col.description = ""
-        if not b_col:
-            b_col = ColumnSpec(  # pragma: no cover
-                name=col_name,
-                display_name=labelize(col_name),
-                type=a_col.type if a_col else "",
-                attributes=[],
-            )
-        if not b_col.display_name:
-            b_col.display_name = labelize(col_name)
-        if not b_col.description:  # pragma: no cover
-            b_col.description = ""
-        if not compare_types:
-            a_col.type = ""
-            b_col.type = ""
-        # Remove primary_key from copies for comparison
-        if "primary_key" in a_col.attributes:
-            a_col.attributes = list(set(a_col.attributes) - {"primary_key"})
-        if "primary_key" in b_col.attributes:
-            b_col.attributes = list(set(b_col.attributes) - {"primary_key"})
+    for col_name in sorted(set(a_map).union(b_map)):
+        a_col = _normalized_column(
+            a_map.get(col_name),
+            col_name,
+            b_map[col_name].type if col_name in b_map else "",
+            compare_types,
+        )
+        b_col = _normalized_column(
+            b_map.get(col_name),
+            col_name,
+            a_map[col_name].type if col_name in a_map else "",
+            compare_types,
+        )
         a_cols.append(a_col)
         b_cols.append(b_col)
     return a_cols == b_cols
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index c185b06d0..ed83bd2a6 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -1,5 +1,9 @@
 import json
+import os
+import subprocess
+import sys
 from datetime import date
+from typing import Any, ClassVar
 
 import pytest
 from pydantic import ValidationError
@@ -26,6 +30,7 @@
     PartitionSpec,
     PartitionType,
     PreAggSpec,
+    SemanticFingerprint,
     SourceSpec,
     TagSpec,
     TransformSpec,
@@ -1629,3 +1634,441 @@ def test_a_schema_namespace_outside_the_deployment_is_rejected(outside):
             ],
         )
     assert "not 'shared' or beneath it" in str(exc_info.value)
+
+
+def semantic_specs() -> dict[str, NodeSpec]:
+    """Representative inputs for each concrete node type."""
+    return {
+        "source": SourceSpec(
+            namespace="analytics",
+            name="orders",
+            catalog="warehouse",
+            schema="sales",
+            table="orders",
+            columns=[ColumnSpec(name="order_id", type="bigint")],
+            primary_key=["order_id"],
+        ),
+        "transform": TransformSpec(
+            namespace="analytics",
+            name="clean_orders",
+            query=(
+                "SELECT order_id AS id, amount FROM ${prefix}orders WHERE amount > 0"
+            ),
+        ),
+        "dimension": DimensionSpec(
+            namespace="analytics",
+            name="order",
+            query="SELECT order_id, status FROM ${prefix}orders",
+        ),
+        "metric": MetricSpec(
+            namespace="analytics",
+            name="total_amount",
+            query="SELECT SUM(amount) AS value FROM ${prefix}orders",
+            required_dimensions=["${prefix}order.status"],
+        ),
+        "cube": CubeSpec(
+            namespace="analytics",
+            name="order_cube",
+            metrics=["${prefix}total_amount", "${prefix}order_count"],
+            dimensions=["${prefix}order.status", "${prefix}order.order_id"],
+            filters=["${prefix}order.status != 'cancelled'", "amount > 0"],
+            columns=[
+                ColumnSpec(
+                    name="${prefix}order.status",
+                    partition=PartitionSpec(type=PartitionType.CATEGORICAL),
+                ),
+            ],
+        ),
+    }
+
+
+def fingerprint(spec: NodeSpec) -> SemanticFingerprint:
+    return spec.semantic_fingerprint()
+
+
+GOLDEN_FINGERPRINTS = {
+    "source": "6f152da51c2d66ca93b9a5cf2e4f828555faeb98210ad58285c726ed81116661",
+    "transform": "98ed2d10f7f273a7ca73ce50777ae12fad28167ab016c01813784e3d90844062",
+    "dimension": "5051483b7d5cbb999e3617deee2d7dfa80764579d6fa2946b0bd407775db3156",
+    "metric": "a1977ae081d398f5667d34f2b34650e263a4b6507eb7c53570f440bbe08f6f4e",
+    "cube": "cbdbac6e9015d0d1c45502073925f9fafe119f9d3adce2ea864d8d4f9ec17628",
+}
+
+
+@pytest.mark.parametrize("node_type", GOLDEN_FINGERPRINTS)
+def test_semantic_fingerprint_golden_digests(node_type):
+    spec = semantic_specs()[node_type]
+    result = fingerprint(spec)
+    assert result == SemanticFingerprint(digest=GOLDEN_FINGERPRINTS[node_type])
+    assert result == fingerprint(spec)
+
+
+def test_semantic_fingerprint_is_independent_of_python_hash_seed():
+    script = """
+from datajunction_server.api.main import app
+from datajunction_server.models.deployment import ColumnSpec, SourceSpec
+spec = SourceSpec(name="s", catalog="c", schema_="s", table="t",
+    columns=[ColumnSpec(name="id", attributes=["z", "primary_key", "a"])])
+print(spec.semantic_fingerprint().digest)
+"""
+
+    def digest_for(seed):
+        return subprocess.check_output(
+            [sys.executable, "-c", script],
+            env={**os.environ, "PYTHONHASHSEED": seed},
+            text=True,
+        ).splitlines()[-1]
+
+    assert digest_for("1") == digest_for("42")
+
+
+def test_semantic_fingerprint_normalizes_empty_and_resolved_source_columns():
+    common = {"name": "source", "catalog": "c", "schema_": "s", "table": "t"}
+    unspecified = SourceSpec(**common, columns=None)
+    empty = SourceSpec(**common, columns=[])
+    columns = [ColumnSpec(name="id", type="bigint")]
+    resolved = SourceSpec(**common, columns=columns)
+    duplicated = SourceSpec(**common, columns=[*columns, columns[0].model_copy()])
+
+    assert fingerprint(unspecified) == fingerprint(empty)
+    assert unspecified.semantic_fingerprint(
+        resolved_columns=columns,
+    ) == fingerprint(resolved)
+    assert fingerprint(resolved) == fingerprint(duplicated)
+    assert fingerprint(
+        CubeSpec(name="cube", metrics=[], dimensions=[], filters=None),
+    ) == fingerprint(CubeSpec(name="cube", metrics=[], dimensions=[], filters=[]))
+
+
+def test_semantic_fingerprint_mapping_order_is_stable():
+    class CanonicalSpec(NodeSpec):
+        semantic_mapping: dict[str, Any]
+        values: list[int]
+        FIELD_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
+            "semantic_mapping": ChangeTier.MAJOR,
+            "values": ChangeTier.MAJOR,
+        }
+        FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
+            "values": ChangeTier.MAJOR,
+        }
+
+    common = {"name": "mapped", "node_type": NodeType.SOURCE, "values": [1, 2]}
+    first = CanonicalSpec(
+        **common,
+        semantic_mapping={"outer": {"a": 1, "b": 2}, "value": 3},
+    )
+    second = CanonicalSpec(
+        **common,
+        semantic_mapping={"value": 3, "outer": {"b": 2, "a": 1}},
+    )
+    assert fingerprint(first) == fingerprint(second)
+    with pytest.raises(TypeError, match="string keys"):
+        fingerprint(first.model_copy(update={"semantic_mapping": {1: "value"}}))
+    assert fingerprint(first) != fingerprint(
+        first.model_copy(update={"values": [2, 1]}),
+    )
+    with pytest.raises(TypeError, match="Unsupported"):
+        fingerprint(first.model_copy(update={"semantic_mapping": {"bad": object()}}))
+    with pytest.raises(ValueError, match="must be finite"):
+        fingerprint(
+            first.model_copy(update={"semantic_mapping": {"bad": float("nan")}}),
+        )
+
+
+def test_semantic_fingerprint_renders_prefixes_and_canonicalizes_sql():
+    from datajunction_server.models.dialect import Dialect
+    from datajunction_server.sql.parsing.ast import render_for_dialect
+
+    parameterized = TransformSpec(
+        namespace="analytics",
+        name="orders",
+        query="SELECT\n id AS order_id\nFROM ${prefix}raw_orders",
+    )
+    rendered = TransformSpec(
+        namespace="analytics",
+        name="orders",
+        query="SELECT id AS order_id FROM analytics.raw_orders",
+    )
+    assert parameterized.query_ast.compare(rendered.query_ast)
+    assert fingerprint(parameterized) == fingerprint(rendered)
+    dialect_query = TransformSpec(
+        name="dialect",
+        query="SELECT COLLECT_LIST(value) AS values FROM source",
+    )
+    dialect_fingerprint = fingerprint(dialect_query)
+    with render_for_dialect(Dialect.TRINO):
+        assert fingerprint(parameterized) == fingerprint(rendered)
+        assert fingerprint(dialect_query) == dialect_fingerprint
+        assert fingerprint(
+            TransformSpec(
+                name="typed",
+                query="SELECT CAST(value AS DECIMAL(10, 2)) FROM source",
+            ),
+        ).digest
+    assert fingerprint(TransformSpec(name="blank", query="")).digest
+    explicit = TransformSpec(name="orders", query="SELECT id AS order_id FROM raw")
+    implicit = TransformSpec(name="orders", query="SELECT id order_id FROM raw")
+    assert not explicit.query_ast.compare(implicit.query_ast)
+    assert fingerprint(explicit) != fingerprint(implicit)
+
+
+def test_canonical_diff_and_fingerprint_share_change_rules():
+    original = TransformSpec(name="node", query="SELECT id AS value FROM source")
+    formatted = TransformSpec(
+        name="node",
+        query=" SELECT  id AS value\nFROM source ",
+    )
+    changed, reordered = original.canonical_diff(formatted)
+    assert (changed, reordered) == ([], [])
+    assert TransformSpec.change_tier(changed, reordered) == ChangeTier.NONE
+    assert fingerprint(original) == fingerprint(formatted)
+
+    source = SourceSpec(
+        name="source",
+        catalog="c",
+        schema_="s",
+        table="t",
+        columns=[ColumnSpec(name="id", type="bigint")],
+    )
+    source_changed = source.model_copy(deep=True)
+    source_changed.columns[0].type = "string"
+    changed, reordered = source.canonical_diff(source_changed)
+    assert (changed, reordered) == (["columns"], [])
+    assert SourceSpec.change_tier(changed, reordered) == ChangeTier.MAJOR
+    assert fingerprint(source) != fingerprint(source_changed)
+    assert source.canonical_diff(original) == (["node_type"], [])
+
+    cube = CubeSpec(name="cube", metrics=["a", "b"], dimensions=[])
+    reordered_cube = cube.model_copy(update={"metrics": ["b", "a", "a"]})
+    changed, reordered = cube.canonical_diff(reordered_cube)
+    assert (changed, reordered) == ([], ["metrics"])
+    assert CubeSpec.change_tier(changed, reordered) == ChangeTier.MINOR
+    assert fingerprint(cube) == fingerprint(reordered_cube)
+
+    legacy_metric = MetricSpec(
+        name="metric",
+        query="SELECT 1",
+        unit="dollar",
+    )
+    structured_metric = MetricSpec(
+        name="metric",
+        query="SELECT 1",
+        direction="neutral",
+        unit={"kind": "currency", "code": "USD"},
+    )
+    assert legacy_metric.canonical_diff(structured_metric) == ([], [])
+    changed, reordered = MetricSpec(
+        name="metric",
+        query="SELECT 1",
+    ).canonical_diff(legacy_metric)
+    assert (changed, reordered) == (["unit_enum"], [])
+    assert MetricSpec.change_tier(changed, reordered) == ChangeTier.MINOR
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("owners", ["other"]),
+        ("display_name", "Orders"),
+        ("description", "Updated description"),
+        ("tags", ["certified"]),
+        ("mode", NodeMode.DRAFT),
+        ("custom_metadata", {"team": "analytics"}),
+    ],
+)
+def test_minor_base_node_fields_preserve_semantic_fingerprint(field, value):
+    original = semantic_specs()["source"]
+    changed = original.model_copy(update={field: value})
+    assert type(original).field_change_tier(field) == ChangeTier.MINOR
+    assert fingerprint(original) == fingerprint(changed)
+
+
+def test_metric_presentation_fields_preserve_semantic_fingerprint():
+    baseline = MetricSpec(name="metric", query="SELECT 1")
+    presentations = [
+        MetricSpec(
+            name="metric",
+            query="SELECT 1",
+            direction="higher_is_better",
+            unit="dollar",
+            significant_digits=3,
+            min_decimal_exponent=-2,
+            max_decimal_exponent=4,
+        ),
+        MetricSpec(
+            name="metric",
+            query="SELECT 1",
+            unit={"kind": "currency", "code": "USD"},
+        ),
+    ]
+    fields = (
+        set(MetricSpec.model_fields)
+        - set(NodeSpec.model_fields)
+        - {
+            "query",
+            "columns",
+            "required_dimensions",
+        }
+    )
+    assert all(
+        MetricSpec.field_change_tier(field) == ChangeTier.MINOR for field in fields
+    )
+    assert all(fingerprint(spec) == fingerprint(baseline) for spec in presentations)
+
+
+@pytest.mark.parametrize(
+    ("node_type", "field", "value"),
+    [
+        ("source", "catalog", "other"),
+        ("source", "schema_", "other"),
+        ("source", "table", "other"),
+        ("source", "primary_key", ["amount"]),
+        ("transform", "query", "SELECT amount FROM analytics.orders"),
+        ("dimension", "query", "SELECT order_id FROM analytics.orders"),
+        ("metric", "query", "SELECT COUNT(*) AS value FROM analytics.orders"),
+        ("metric", "required_dimensions", ["analytics.order.order_id"]),
+        ("cube", "metrics", ["analytics.order_count"]),
+        ("cube", "dimensions", ["analytics.order.order_id"]),
+        ("cube", "filters", ["amount >= 0"]),
+    ],
+)
+def test_major_node_fields_change_semantic_fingerprint(node_type, field, value):
+    original = semantic_specs()[node_type]
+    changed = original.model_copy(update={field: value})
+    assert type(original).field_change_tier(field) == ChangeTier.MAJOR
+    assert fingerprint(original) != fingerprint(changed)
+
+
+def test_semantic_fingerprint_column_rules_match_equality():
+    source = SourceSpec(
+        name="source",
+        catalog="c",
+        schema_="s",
+        table="t",
+        columns=[
+            ColumnSpec(name="id", type="bigint", attributes=["primary_key", "id"]),
+            ColumnSpec(name="value", type="string"),
+        ],
+    )
+    source_reordered = source.model_copy(deep=True)
+    source_reordered.columns = list(reversed(source_reordered.columns or []))
+    source_reordered.columns[1].attributes = ["id", "primary_key"]
+    source_type_changed = source.model_copy(deep=True)
+    source_type_changed.columns[0].type = "integer"
+    assert eq_columns(source.columns, source_reordered.columns)
+    assert fingerprint(source) == fingerprint(source_reordered)
+    assert not eq_columns(source.columns, source_type_changed.columns)
+    assert fingerprint(source) != fingerprint(source_type_changed)
+
+    for spec_class in (TransformSpec, DimensionSpec):
+        original = spec_class(
+            name="derived",
+            query="SELECT id FROM source",
+            columns=[ColumnSpec(name="id", type="bigint")],
+        )
+        inferred_type_changed = original.model_copy(deep=True)
+        inferred_type_changed.columns[0].type = "string"
+        metadata_changed = original.model_copy(deep=True)
+        metadata_changed.columns[0].attributes = ["identifier"]
+        assert eq_columns(original.columns, inferred_type_changed.columns, False)
+        assert fingerprint(original) == fingerprint(inferred_type_changed)
+        assert not eq_columns(original.columns, metadata_changed.columns, False)
+        assert fingerprint(original) != fingerprint(metadata_changed)
+
+
+def test_semantic_fingerprint_dimension_link_rules_match_equality():
+    from datajunction_server.models.dimensionlink import SparkJoinStrategy
+
+    links = [
+        DimensionReferenceLinkSpec(
+            node_column="customer_id",
+            dimension="${prefix}customer.id",
+            role="customer",
+        ),
+        DimensionJoinLinkSpec(
+            dimension_node="${prefix}date",
+            join_on="${prefix}orders.date_id = ${prefix}date.id",
+            role="date",
+        ),
+    ]
+    original = TransformSpec(
+        namespace="analytics",
+        name="orders",
+        query="SELECT 1",
+        dimension_links=links,
+    )
+    reordered = original.model_copy(update={"dimension_links": list(reversed(links))})
+    changed = original.model_copy(deep=True)
+    changed.dimension_links[0].role = "buyer"
+    hint_changed = original.model_copy(
+        update={
+            "dimension_links": [
+                links[0],
+                links[1].model_copy(
+                    update={"spark_hints": SparkJoinStrategy.BROADCAST},
+                ),
+            ],
+        },
+    )
+    assert original == reordered
+    assert fingerprint(original) == fingerprint(reordered)
+    assert original == hint_changed
+    assert fingerprint(original) == fingerprint(hint_changed)
+    assert original != changed
+    assert fingerprint(original) != fingerprint(changed)
+
+
+def test_semantic_fingerprint_normalizes_primary_keys_and_cube_ordering():
+    source = semantic_specs()["source"]
+    assert fingerprint(source) == fingerprint(
+        source.model_copy(
+            update={"primary_key": ["order_id", "order_id"]},
+        ),
+    )
+
+    cube = semantic_specs()["cube"]
+    reordered = cube.model_copy(deep=True)
+    reordered.metrics.reverse()
+    reordered.dimensions.reverse()
+    reordered.filters = list(reversed(reordered.filters or []))
+    assert fingerprint(cube) == fingerprint(reordered)
+    metric = MetricSpec(
+        name="metric",
+        query="SELECT 1",
+        required_dimensions=["one", "two"],
+    )
+    assert fingerprint(metric) == fingerprint(
+        metric.model_copy(update={"required_dimensions": ["two", "one"]}),
+    )
+    assert fingerprint(cube) == fingerprint(
+        cube.model_copy(
+            update={"metrics": [*cube.metrics, "${prefix}order_count"]},
+        ),
+    )
+    assert fingerprint(cube) != fingerprint(
+        cube.model_copy(
+            update={"metrics": [*cube.metrics, "${prefix}average_amount"]},
+        ),
+    )
+    assert fingerprint(cube) != fingerprint(
+        cube.model_copy(
+            update={"filters": ["amount > 1"]},
+        ),
+    )
+    partition_changed = cube.model_copy(deep=True)
+    partition_changed.columns[0].partition.type = PartitionType.TEMPORAL
+    assert fingerprint(cube) != fingerprint(partition_changed)
+
+
+@pytest.mark.parametrize(
+    "digest",
+    ["a" * 63, "a" * 65, "A" * 64, "g" * 64],
+)
+def test_semantic_fingerprint_digest_validation(digest):
+    with pytest.raises(ValidationError):
+        SemanticFingerprint(digest=digest)
+
+
+def test_semantic_fingerprint_rejects_unknown_version():
+    with pytest.raises(ValueError, match="Unsupported semantic fingerprint version: 2"):
+        semantic_specs()["source"].semantic_fingerprint(version=2)

From 3a9b4b58da01df8d32f6b6c27e50e31eeb664911 Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 11:00:18 -0700
Subject: [PATCH 02/11] Compose semantic fingerprints from parent hashes

Use a versioned Merkle-style digest so persisted object metadata changes whenever any semantic ancestor changes.
---
 .../datajunction_server/models/deployment.py  | 18 +++++++-
 .../tests/models/deployment_test.py           | 42 ++++++++++++++++---
 2 files changed, 53 insertions(+), 7 deletions(-)

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index 745bb7f2a..b3d3c4755 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -73,6 +73,7 @@ class ChangeTier(IntEnum):
 class SemanticFingerprint(BaseModel):
     """A digest of a node's semantic definition."""
 
+    version: Literal[1] = 1
     digest: str = Field(
         min_length=64,
         max_length=64,
@@ -725,17 +726,30 @@ def semantic_fingerprint(
         self,
         version: int = _SEMANTIC_FINGERPRINT_VERSION,
         *,
+        parent_fingerprints: Iterable[SemanticFingerprint] = (),
         resolved_columns: list[ColumnSpec] | None = None,
     ) -> SemanticFingerprint:
         """Return a fingerprint of this node's semantic definition."""
         if type(version) is not int or version != _SEMANTIC_FINGERPRINT_VERSION:
             raise ValueError(f"Unsupported semantic fingerprint version: {version}")
-        payload = _semantic_fingerprint_payload(
+        node_payload = _semantic_fingerprint_payload(
             self,
             resolved_columns=resolved_columns,
         )
+        node_digest = hashlib.sha256(
+            _canonical_json(node_payload).encode("utf-8"),
+        ).hexdigest()
+        parents = list(parent_fingerprints)
+        if any(parent.version != version for parent in parents):
+            raise ValueError("Parent fingerprint version does not match node version")
+        payload = {
+            "domain": "datajunction/node-semantic-merkle",
+            "version": version,
+            "node": node_digest,
+            "parents": sorted({parent.digest for parent in parents}),
+        }
         digest = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
-        return SemanticFingerprint(digest=digest)
+        return SemanticFingerprint(version=version, digest=digest)
 
     def canonical_diff(
         self,
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index ed83bd2a6..f35734a6f 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -1687,11 +1687,11 @@ def fingerprint(spec: NodeSpec) -> SemanticFingerprint:
 
 
 GOLDEN_FINGERPRINTS = {
-    "source": "6f152da51c2d66ca93b9a5cf2e4f828555faeb98210ad58285c726ed81116661",
-    "transform": "98ed2d10f7f273a7ca73ce50777ae12fad28167ab016c01813784e3d90844062",
-    "dimension": "5051483b7d5cbb999e3617deee2d7dfa80764579d6fa2946b0bd407775db3156",
-    "metric": "a1977ae081d398f5667d34f2b34650e263a4b6507eb7c53570f440bbe08f6f4e",
-    "cube": "cbdbac6e9015d0d1c45502073925f9fafe119f9d3adce2ea864d8d4f9ec17628",
+    "source": "71dcbc388988c2bdd850670427710384687b58565ee38ca392dc220adfed868d",
+    "transform": "797af072eb15831df786149716171f0a400af7f35f6031f8eee3939733cab9c4",
+    "dimension": "965282bbe90a4fcc66576fba3550df42c09ff4dc51be2ee433a3897d48026edb",
+    "metric": "738703ffa80a72b00fc829697cf259aac317dc34290cb1aae4c0678a34f3a948",
+    "cube": "9b0a56d974d1e3769bc2db94e2cfbae7a6a4839f664eebd2a4387ef112ceea81",
 }
 
 
@@ -1701,6 +1701,7 @@ def test_semantic_fingerprint_golden_digests(node_type):
     result = fingerprint(spec)
     assert result == SemanticFingerprint(digest=GOLDEN_FINGERPRINTS[node_type])
     assert result == fingerprint(spec)
+    assert result.version == 1
 
 
 def test_semantic_fingerprint_is_independent_of_python_hash_seed():
@@ -2070,5 +2071,36 @@ def test_semantic_fingerprint_digest_validation(digest):
 
 
 def test_semantic_fingerprint_rejects_unknown_version():
+    with pytest.raises(ValidationError):
+        SemanticFingerprint(version=2, digest="a" * 64)
     with pytest.raises(ValueError, match="Unsupported semantic fingerprint version: 2"):
         semantic_specs()["source"].semantic_fingerprint(version=2)
+
+
+def test_semantic_fingerprint_combines_sorted_parent_hashes():
+    node = TransformSpec(name="node", query="SELECT id FROM parent")
+    first = SourceSpec(name="first", catalog="c", schema_="s", table="first")
+    second = SourceSpec(name="second", catalog="c", schema_="s", table="second")
+    first_hash = fingerprint(first)
+    second_hash = fingerprint(second)
+
+    expected = node.semantic_fingerprint(
+        parent_fingerprints=[first_hash, second_hash],
+    )
+    assert expected == node.semantic_fingerprint(
+        parent_fingerprints=[second_hash, first_hash, first_hash],
+    )
+    assert expected != node.semantic_fingerprint(
+        parent_fingerprints=[
+            first_hash,
+            SourceSpec(
+                name="second",
+                catalog="c",
+                schema_="s",
+                table="changed",
+            ).semantic_fingerprint(),
+        ],
+    )
+    mismatched = SemanticFingerprint.model_construct(version=2, digest="b" * 64)
+    with pytest.raises(ValueError, match="Parent fingerprint version"):
+        node.semantic_fingerprint(parent_fingerprints=[mismatched])

From f7b7caf1a67824e5ecf48b739a9bb90906833aed Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 12:05:10 -0700
Subject: [PATCH 03/11] Harden fingerprint extension contracts

Require explicit list-order tiers, dispatch immutable fingerprint versions, and route link equality through the shared comparison key.
---
 .../datajunction_server/models/deployment.py  | 120 +++++++++++++-----
 .../tests/models/deployment_test.py           |  10 +-
 2 files changed, 94 insertions(+), 36 deletions(-)

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index b3d3c4755..69bffb1cd 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -1,7 +1,7 @@
 import hashlib
 import json
 import math
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from enum import Enum, IntEnum
 from typing import Annotated, Any, ClassVar, Literal, get_args, get_origin
 
@@ -73,15 +73,22 @@ class ChangeTier(IntEnum):
 class SemanticFingerprint(BaseModel):
     """A digest of a node's semantic definition."""
 
-    version: Literal[1] = 1
+    version: int = 1
     digest: str = Field(
         min_length=64,
         max_length=64,
         pattern=r"^[0-9a-f]+$",
     )
 
+    @field_validator("version")
+    @classmethod
+    def validate_version(cls, version: int) -> int:
+        if version not in _SEMANTIC_FINGERPRINT_BUILDERS:
+            raise ValueError(f"Unsupported semantic fingerprint version: {version}")
+        return version
+
 
-_SEMANTIC_FINGERPRINT_VERSION = 1
+_LATEST_SEMANTIC_FINGERPRINT_VERSION = 1
 
 
 def fold_change_tiers(tiers: Iterable[ChangeTier]) -> ChangeTier:
@@ -549,11 +556,6 @@ def _comparison_key(self) -> tuple[Any, ...]:
             self.default_value,
         )
 
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, DimensionJoinLinkSpec):
-            return False  # pragma: no cover
-        return self._comparison_key() == other._comparison_key()
-
 
 class DimensionReferenceLinkSpec(DimensionLinkSpec):
     """
@@ -591,11 +593,6 @@ def _comparison_key(self) -> tuple[Any, ...]:
             self.node_column,
         )
 
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, DimensionReferenceLinkSpec):
-            return False
-        return self._comparison_key() == other._comparison_key()
-
 
 def render_prefixes(parameterized_string: str, prefix: str | None = None) -> str:
     """
@@ -648,7 +645,10 @@ class NodeSpec(NamespacedSpec):
     # anything. Fields absent here are not order-sensitive, so reordering them is
     # not a change at all. `diff()` compares list fields as sets and cannot see a
     # reorder on its own, which is why `order_diff()` exists alongside it.
-    FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {}
+    FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
+        "owners": ChangeTier.NONE,
+        "tags": ChangeTier.NONE,
+    }
 
     _query_ast: Any | None = PrivateAttr(default=None)
     # Internal: marks specs from already-validated sources (e.g., branch copies)
@@ -724,32 +724,20 @@ def rendered_spec(self) -> "NodeSpec":
 
     def semantic_fingerprint(
         self,
-        version: int = _SEMANTIC_FINGERPRINT_VERSION,
+        version: int = _LATEST_SEMANTIC_FINGERPRINT_VERSION,
         *,
         parent_fingerprints: Iterable[SemanticFingerprint] = (),
         resolved_columns: list[ColumnSpec] | None = None,
     ) -> SemanticFingerprint:
         """Return a fingerprint of this node's semantic definition."""
-        if type(version) is not int or version != _SEMANTIC_FINGERPRINT_VERSION:
+        builder = _SEMANTIC_FINGERPRINT_BUILDERS.get(version)
+        if builder is None:
             raise ValueError(f"Unsupported semantic fingerprint version: {version}")
-        node_payload = _semantic_fingerprint_payload(
+        return builder(
             self,
+            parent_fingerprints,
             resolved_columns=resolved_columns,
         )
-        node_digest = hashlib.sha256(
-            _canonical_json(node_payload).encode("utf-8"),
-        ).hexdigest()
-        parents = list(parent_fingerprints)
-        if any(parent.version != version for parent in parents):
-            raise ValueError("Parent fingerprint version does not match node version")
-        payload = {
-            "domain": "datajunction/node-semantic-merkle",
-            "version": version,
-            "node": node_digest,
-            "parents": sorted({parent.digest for parent in parents}),
-        }
-        digest = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
-        return SemanticFingerprint(version=version, digest=digest)
 
     def canonical_diff(
         self,
@@ -818,6 +806,11 @@ def has_explicit_change_tier(cls, field: str) -> bool:
         """Whether some class in the MRO classifies `field`."""
         return cls._declared_tier("FIELD_CHANGE_TIERS", field) is not None
 
+    @classmethod
+    def has_explicit_order_change_tier(cls, field: str) -> bool:
+        """Whether some class in the MRO classifies reordering `field`."""
+        return cls._declared_tier("FIELD_ORDER_CHANGE_TIERS", field) is not None
+
     @classmethod
     def unclassified_fields(cls) -> list[str]:
         """Fields on this spec class that nobody classified. Should always be empty."""
@@ -827,13 +820,27 @@ def unclassified_fields(cls) -> list[str]:
             if not cls.has_explicit_change_tier(field)
         ]
 
+    @classmethod
+    def unclassified_list_order_fields(cls) -> list[str]:
+        """List fields without an explicit order classification."""
+        return [
+            field
+            for field, field_info in cls.model_fields.items()
+            if _annotation_contains_list(field_info.annotation)
+            and not cls.has_explicit_order_change_tier(field)
+        ]
+
     @classmethod
     def order_sensitive_fields(cls) -> list[str]:
-        """Fields for which some class in the MRO classifies a reorder."""
+        """Fields whose declared reorder tier is not NONE."""
         fields: dict[str, None] = {}
         for klass in cls.__mro__:
-            for field in klass.__dict__.get("FIELD_ORDER_CHANGE_TIERS", {}):
-                fields.setdefault(field, None)
+            for field, tier in klass.__dict__.get(
+                "FIELD_ORDER_CHANGE_TIERS",
+                {},
+            ).items():
+                if tier != ChangeTier.NONE:
+                    fields.setdefault(field, None)
         return list(fields)
 
     @classmethod
@@ -892,6 +899,11 @@ class LinkableNodeSpec(NodeSpec):
         "dimension_links": ChangeTier.MAJOR,
         "primary_key": ChangeTier.MAJOR,
     }
+    FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
+        "columns": ChangeTier.NONE,
+        "dimension_links": ChangeTier.NONE,
+        "primary_key": ChangeTier.NONE,
+    }
 
     @model_validator(mode="after")
     def set_namespaces(self):
@@ -1046,6 +1058,10 @@ class MetricSpec(NodeSpec):
         "min_decimal_exponent": ChangeTier.MINOR,
         "max_decimal_exponent": ChangeTier.MINOR,
     }
+    FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
+        "columns": ChangeTier.NONE,
+        "required_dimensions": ChangeTier.NONE,
+    }
 
     # Class-level adapter used by __init__ to eagerly validate structured
     # unit input. `ClassVar` keeps Pydantic from treating it as a field.
@@ -1227,6 +1243,8 @@ class CubeSpec(NodeSpec):
         # Filters are ANDed together, so their ordering carries no meaning at all
         # and reordering them is genuinely a no-op.
         "filters": ChangeTier.NONE,
+        "columns": ChangeTier.NONE,
+        "materialization": ChangeTier.NONE,
     }
 
     @field_validator("materialization", mode="before")
@@ -1596,6 +1614,40 @@ def _semantic_fingerprint_payload(
     }
 
 
+def _semantic_fingerprint_v1(
+    spec: NodeSpec,
+    parent_fingerprints: Iterable[SemanticFingerprint],
+    *,
+    resolved_columns: list[ColumnSpec] | None,
+) -> SemanticFingerprint:
+    node_payload = _semantic_fingerprint_payload(
+        spec,
+        resolved_columns=resolved_columns,
+    )
+    node_digest = hashlib.sha256(
+        _canonical_json(node_payload).encode("utf-8"),
+    ).hexdigest()
+    parents = list(parent_fingerprints)
+    if any(parent.version != 1 for parent in parents):
+        raise ValueError("Parent fingerprint version does not match node version")
+    payload = {
+        "domain": "datajunction/node-semantic-merkle",
+        "version": 1,
+        "node": node_digest,
+        "parents": sorted({parent.digest for parent in parents}),
+    }
+    digest = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
+    return SemanticFingerprint(version=1, digest=digest)
+
+
+_SEMANTIC_FINGERPRINT_BUILDERS: dict[
+    int,
+    Callable[..., SemanticFingerprint],
+] = {
+    1: _semantic_fingerprint_v1,
+}
+
+
 def _canonical_node_diff(
     one: NodeSpec,
     two: NodeSpec,
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index f35734a6f..5b0fd4dfa 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -890,6 +890,12 @@ def test_every_spec_field_has_an_explicit_change_tier():
         if spec_class.unclassified_fields()
     }
     assert unclassified == {}
+    unclassified_order = {
+        spec_class.__name__: spec_class.unclassified_list_order_fields()
+        for spec_class in all_node_spec_classes()
+        if spec_class.unclassified_list_order_fields()
+    }
+    assert unclassified_order == {}
 
 
 def test_change_tier_lookup_walks_the_mro():
@@ -911,7 +917,7 @@ def test_change_tier_lookup_walks_the_mro():
     assert TransformSpec.field_change_tier("primary_key") == ChangeTier.MAJOR
     assert TransformSpec.field_change_tier("tags") == ChangeTier.MINOR
     assert TransformSpec.order_sensitive_fields() == []
-    assert CubeSpec.order_sensitive_fields() == ["metrics", "dimensions", "filters"]
+    assert CubeSpec.order_sensitive_fields() == ["metrics", "dimensions"]
 
 
 def test_fold_change_tiers():
@@ -1000,7 +1006,7 @@ def test_cube_spec_order_diff():
     assert one.order_diff(
         a_cube(dimensions=["ns.d.two", "ns.d.one"], filters=["x = 1", "y = 2"]),
     ) == ["dimensions"]
-    assert one.order_diff(a_cube(filters=["y = 2", "x = 1"])) == ["filters"]
+    assert one.order_diff(a_cube(filters=["y = 2", "x = 1"])) == []
     # A set change is not a reorder — diff() reports that one instead.
     assert one.order_diff(a_cube(metrics=["ns.a"], filters=["x = 1", "y = 2"])) == []
     assert one.diff(a_cube(metrics=["ns.a"], filters=["x = 1", "y = 2"])) == ["metrics"]

From a1f6897a611882135873644a297e0b562664b24c Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 12:29:01 -0700
Subject: [PATCH 04/11] Update the cube reorder assertion

Reflect that NONE-tier filter ordering is omitted from the deployment changelog while the dimension reorder remains minor.
---
 datajunction-server/tests/api/deployments_test.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/datajunction-server/tests/api/deployments_test.py b/datajunction-server/tests/api/deployments_test.py
index a2f5b161b..006652bb2 100644
--- a/datajunction-server/tests/api/deployments_test.py
+++ b/datajunction-server/tests/api/deployments_test.py
@@ -2649,9 +2649,7 @@ async def test_deploy_cube_dimension_reorder(
         assert data["status"] == "success"
         assert data["results"][-1] == {
             "deploy_type": "node",
-            # The filter reorder is reported for the reader's benefit but earns no
-            # version of its own: v1.1 comes from the dimension reorder alone.
-            "message": "Updated cube (v1.1)\n└─ Reordered dimensions, filters",
+            "message": "Updated cube (v1.1)\n└─ Reordered dimensions",
             "name": f"{namespace}.default.repairs_cube",
             "operation": "update",
             "changed_fields": [],

From db3042a84f054cf20825273139d3ce76ca5a2f0e Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 15:35:18 -0700
Subject: [PATCH 05/11] Isolate and freeze semantic fingerprint construction

Move canonicalization and structural SQL serialization behind a dedicated package while preserving version 1 digests and aligning metric field semantics.
---
 .../datajunction_server/models/deployment.py  | 426 +++---------------
 .../models/semantic_fingerprint.py            |  27 ++
 .../semantic_fingerprints/__init__.py         |   8 +
 .../semantic_fingerprints/canonical.py        | 270 +++++++++++
 .../semantic_fingerprints/engine.py           |  55 +++
 .../semantic_fingerprints/v1.py               | 106 +++++
 .../sql/parsing/structural.py                 | 137 ++++++
 .../tests/models/deployment_test.py           |  88 ++--
 .../tests/sql/parsing/test_structural.py      |  32 ++
 9 files changed, 740 insertions(+), 409 deletions(-)
 create mode 100644 datajunction-server/datajunction_server/models/semantic_fingerprint.py
 create mode 100644 datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
 create mode 100644 datajunction-server/datajunction_server/semantic_fingerprints/canonical.py
 create mode 100644 datajunction-server/datajunction_server/semantic_fingerprints/engine.py
 create mode 100644 datajunction-server/datajunction_server/semantic_fingerprints/v1.py
 create mode 100644 datajunction-server/datajunction_server/sql/parsing/structural.py
 create mode 100644 datajunction-server/tests/sql/parsing/test_structural.py

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index 69bffb1cd..1564bf374 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -1,9 +1,7 @@
-import hashlib
 import json
-import math
-from collections.abc import Callable, Iterable
+from collections.abc import Iterable
 from enum import Enum, IntEnum
-from typing import Annotated, Any, ClassVar, Literal, get_args, get_origin
+from typing import Annotated, Any, ClassVar, Literal
 
 from pydantic import (
     AliasChoices,
@@ -21,7 +19,6 @@
     DJInvalidDeploymentConfig,
     DJInvalidInputException,
 )
-from datajunction_server.models.base import labelize
 from datajunction_server.models.dimensionlink import (
     JoinCardinality,
     JoinType,
@@ -43,6 +40,10 @@
     NodeType,
 )
 from datajunction_server.models.partition import Granularity, PartitionType
+from datajunction_server.models.semantic_fingerprint import (
+    LATEST_SEMANTIC_FINGERPRINT_VERSION,
+    SemanticFingerprint,
+)
 from datajunction_server.models.unit import (
     Unit,
     legacy_unit_to_structured,
@@ -70,27 +71,6 @@ class ChangeTier(IntEnum):
     MAJOR = 20
 
 
-class SemanticFingerprint(BaseModel):
-    """A digest of a node's semantic definition."""
-
-    version: int = 1
-    digest: str = Field(
-        min_length=64,
-        max_length=64,
-        pattern=r"^[0-9a-f]+$",
-    )
-
-    @field_validator("version")
-    @classmethod
-    def validate_version(cls, version: int) -> int:
-        if version not in _SEMANTIC_FINGERPRINT_BUILDERS:
-            raise ValueError(f"Unsupported semantic fingerprint version: {version}")
-        return version
-
-
-_LATEST_SEMANTIC_FINGERPRINT_VERSION = 1
-
-
 def fold_change_tiers(tiers: Iterable[ChangeTier]) -> ChangeTier:
     """
     Reduce the tiers of several individual changes to the tier of the change as a
@@ -649,7 +629,6 @@ class NodeSpec(NamespacedSpec):
         "owners": ChangeTier.NONE,
         "tags": ChangeTier.NONE,
     }
-
     _query_ast: Any | None = PrivateAttr(default=None)
     # Internal: marks specs from already-validated sources (e.g., branch copies)
     # that can skip expensive SQL parsing and validation
@@ -724,18 +703,18 @@ def rendered_spec(self) -> "NodeSpec":
 
     def semantic_fingerprint(
         self,
-        version: int = _LATEST_SEMANTIC_FINGERPRINT_VERSION,
+        version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION,
         *,
         parent_fingerprints: Iterable[SemanticFingerprint] = (),
         resolved_columns: list[ColumnSpec] | None = None,
     ) -> SemanticFingerprint:
         """Return a fingerprint of this node's semantic definition."""
-        builder = _SEMANTIC_FINGERPRINT_BUILDERS.get(version)
-        if builder is None:
-            raise ValueError(f"Unsupported semantic fingerprint version: {version}")
-        return builder(
+        from datajunction_server.semantic_fingerprints import fingerprint_node
+
+        return fingerprint_node(
             self,
-            parent_fingerprints,
+            version,
+            parent_fingerprints=parent_fingerprints,
             resolved_columns=resolved_columns,
         )
 
@@ -747,7 +726,11 @@ def canonical_diff(
         other_resolved_columns: list[ColumnSpec] | None = None,
     ) -> tuple[list[str], list[str]]:
         """Compare two specs using the same canonical values as fingerprints."""
-        return _canonical_node_diff(
+        from datajunction_server.semantic_fingerprints import (
+            canonical_diff as compare_semantics,
+        )
+
+        return compare_semantics(
             self,
             other,
             resolved_columns=resolved_columns,
@@ -823,10 +806,14 @@ def unclassified_fields(cls) -> list[str]:
     @classmethod
     def unclassified_list_order_fields(cls) -> list[str]:
         """List fields without an explicit order classification."""
+        from datajunction_server.semantic_fingerprints.canonical import (
+            annotation_contains_list,
+        )
+
         return [
             field
             for field, field_info in cls.model_fields.items()
-            if _annotation_contains_list(field_info.annotation)
+            if annotation_contains_list(field_info.annotation)
             and not cls.has_explicit_order_change_tier(field)
         ]
 
@@ -926,6 +913,10 @@ def links_mapping(self) -> dict[tuple[str, str | None], DimensionLinkSpec]:
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, LinkableNodeSpec):
             return False  # pragma: no cover
+        from datajunction_server.semantic_fingerprints.canonical import (
+            canonical_dimension_links,
+        )
+
         return (
             super().__eq__(other)
             and eq_columns(
@@ -933,11 +924,11 @@ def __eq__(self, other: object) -> bool:
                 other.columns,
                 compare_types=self.node_type == NodeType.SOURCE,
             )
-            and _canonical_dimension_links(
+            and canonical_dimension_links(
                 self.dimension_links,
                 preserve_order=False,
             )
-            == _canonical_dimension_links(
+            == canonical_dimension_links(
                 other.dimension_links,
                 preserve_order=False,
             )
@@ -1045,7 +1036,7 @@ class MetricSpec(NodeSpec):
 
     FIELD_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
         "query": ChangeTier.MAJOR,
-        "columns": ChangeTier.MAJOR,
+        "columns": ChangeTier.NONE,
         # Required dimensions constrain which queries the metric can answer.
         "required_dimensions": ChangeTier.MAJOR,
         # Everything below is presentation metadata on the metric's single output
@@ -1161,10 +1152,21 @@ def _canonical_unit(self) -> "Unit | None":
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, MetricSpec):
             return False
+        from datajunction_server.semantic_fingerprints.canonical import (
+            canonical_sequence,
+        )
+
         return (
             super().__eq__(other)
             and self.query_ast.compare(other.query_ast)
-            and (self.required_dimensions or []) == (other.required_dimensions or [])
+            and canonical_sequence(
+                self.rendered_required_dimensions,
+                preserve_order=False,
+            )
+            == canonical_sequence(
+                other.rendered_required_dimensions,
+                preserve_order=False,
+            )
             and eq_or_fallback(self.direction, other.direction, MetricDirection.NEUTRAL)
             and self._canonical_unit() == other._canonical_unit()
             and self.significant_digits == other.significant_digits
@@ -1366,9 +1368,13 @@ def __eq__(self, other: object) -> bool:
 
         # Compare only partition config for user-specified columns.
         # Cube element columns (types, order, attributes) are auto-derived and ignored.
-        return _canonical_cube_columns(
+        from datajunction_server.semantic_fingerprints.canonical import (
+            canonical_cube_columns,
+        )
+
+        return canonical_cube_columns(
             self.rendered_columns,
-        ) == _canonical_cube_columns(other.rendered_columns)
+        ) == canonical_cube_columns(other.rendered_columns)
 
 
 NodeUnion = Annotated[
@@ -1377,338 +1383,6 @@ def __eq__(self, other: object) -> bool:
 ]
 
 
-def _canonical_json(value: Any) -> str:
-    """Serialize a fingerprint value deterministically."""
-    return json.dumps(
-        value,
-        allow_nan=False,
-        ensure_ascii=False,
-        separators=(",", ":"),
-        sort_keys=True,
-    )
-
-
-def _canonical_value(value: Any) -> Any:
-    """Convert supported values to deterministic JSON-compatible values."""
-    if isinstance(value, Enum):
-        return _canonical_value(value.value)
-    if isinstance(value, BaseModel):
-        return _canonical_value(value.model_dump(mode="python"))
-    if isinstance(value, dict):
-        if any(not isinstance(key, str) for key in value):
-            raise TypeError("Semantic fingerprint mappings require string keys")
-        return {key: _canonical_value(item) for key, item in value.items()}
-    if isinstance(value, (list, tuple)):
-        return [_canonical_value(item) for item in value]
-    if isinstance(value, float) and not math.isfinite(value):
-        raise ValueError("Semantic fingerprint values must be finite")
-    if value is None or isinstance(value, (str, int, float, bool)):
-        return value
-    raise TypeError(
-        f"Unsupported semantic fingerprint value: {type(value).__name__}",
-    )
-
-
-def _canonical_sql_ast(query_ast: Any) -> Any:
-    """Serialize parsed SQL from AST fields without invoking SQL rendering."""
-    from datajunction_server.sql.parsing.ast import Node as SQLNode
-    from datajunction_server.sql.parsing.types import ColumnType
-
-    def serialize(value: Any) -> Any:
-        if isinstance(value, SQLNode):
-            return {
-                "type": f"{type(value).__module__}.{type(value).__qualname__}",
-                "fields": {
-                    name: serialize(field_value)
-                    for name, field_value in value.fields(
-                        flat=False,
-                        nodes_only=False,
-                        obfuscated=False,
-                        nones=True,
-                        named=True,
-                    )
-                },
-            }
-        if isinstance(value, ColumnType):
-            return {
-                "type": f"{type(value).__module__}.{type(value).__qualname__}",
-                "value": str(value),
-            }
-        if isinstance(value, Enum):
-            return {
-                "type": f"{type(value).__module__}.{type(value).__qualname__}",
-                "value": serialize(value.value),
-            }
-        if isinstance(value, (list, tuple)):
-            return [serialize(item) for item in value]
-        return _canonical_value(value)
-
-    return serialize(query_ast)
-
-
-def _canonical_sequence(
-    values: Iterable[Any],
-    *,
-    preserve_order: bool,
-) -> list[Any]:
-    """Canonicalize a sequence while removing duplicate semantic values."""
-    unique: dict[str, Any] = {}
-    for value in values:
-        canonical = _canonical_value(value)
-        unique.setdefault(_canonical_json(canonical), canonical)
-    return (
-        list(unique.values())
-        if preserve_order
-        else [unique[key] for key in sorted(unique)]
-    )
-
-
-def _annotation_contains_list(annotation: Any) -> bool:
-    return get_origin(annotation) is list or any(
-        _annotation_contains_list(argument) for argument in get_args(annotation)
-    )
-
-
-def _normalized_column(
-    column: ColumnSpec | None,
-    name: str,
-    fallback_type: str | None,
-    compare_types: bool,
-) -> ColumnSpec:
-    normalized = (
-        column.model_copy()
-        if column
-        else ColumnSpec(
-            name=name,
-            display_name=labelize(name),
-            type=fallback_type or "",
-            attributes=[],
-        )
-    )
-    normalized.display_name = normalized.display_name or labelize(name)
-    normalized.description = normalized.description or ""
-    normalized.attributes = sorted(set(normalized.attributes) - {"primary_key"})
-    if not compare_types:
-        normalized.type = ""
-    return normalized
-
-
-def _canonical_columns(
-    columns: list[ColumnSpec] | None,
-    *,
-    compare_types: bool,
-) -> list[Any]:
-    column_map = {column.name: column for column in columns or []}
-    canonical = []
-    for name in sorted(column_map):
-        column = column_map[name]
-        normalized = _normalized_column(column, name, column.type, compare_types)
-        if not compare_types:
-            default = _normalized_column(None, name, column.type, compare_types)
-            if normalized == default:
-                continue
-        canonical.append(_canonical_value(normalized))
-    return canonical
-
-
-def _canonical_dimension_links(
-    links: list[DimensionJoinLinkSpec | DimensionReferenceLinkSpec] | None,
-    *,
-    preserve_order: bool,
-) -> list[Any]:
-    return _canonical_sequence(
-        (link._comparison_key() for link in links or []),
-        preserve_order=preserve_order,
-    )
-
-
-def _canonical_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]:
-    return {
-        column.name: _canonical_value(column.partition)
-        for column in columns or []
-        if column.partition
-    }
-
-
-def _canonical_field_value(
-    spec: NodeSpec,
-    field: str,
-    *,
-    resolved_columns: list[ColumnSpec] | None = None,
-    preserve_order: bool = False,
-) -> Any:
-    value = getattr(spec, field)
-    if field == "query":
-        return (
-            _canonical_sql_ast(spec.query_ast)
-            if spec.query_ast is not None
-            else spec.rendered_query
-        )
-    if field == "columns":
-        if isinstance(spec, CubeSpec):
-            return _canonical_cube_columns(spec.rendered_columns)
-        return _canonical_columns(
-            resolved_columns
-            if isinstance(spec, SourceSpec) and resolved_columns is not None
-            else value,
-            compare_types=isinstance(spec, SourceSpec),
-        )
-    if field == "dimension_links" and isinstance(spec, LinkableNodeSpec):
-        return _canonical_dimension_links(
-            spec.dimension_links,
-            preserve_order=preserve_order,
-        )
-    if field == "unit_enum" and isinstance(spec, MetricSpec):
-        return _canonical_value(spec._canonical_unit())
-    if field == "direction" and isinstance(spec, MetricSpec):
-        value = value or MetricDirection.NEUTRAL
-    if field == "description":
-        value = value or None
-    if field == "custom_metadata":
-        value = value or {}
-    if value is None and _annotation_contains_list(
-        type(spec).model_fields[field].annotation,
-    ):
-        value = []
-
-    canonical = _canonical_value(value)
-    return (
-        _canonical_sequence(canonical, preserve_order=preserve_order)
-        if isinstance(canonical, list)
-        else canonical
-    )
-
-
-def _semantic_fingerprint_payload(
-    spec: NodeSpec,
-    *,
-    resolved_columns: list[ColumnSpec] | None,
-) -> dict[str, Any]:
-    for field, field_info in type(spec).model_fields.items():
-        if field in {"name", "namespace", "node_type"} or field_info.exclude is True:
-            continue
-        if type(spec).field_change_tier(field) == ChangeTier.MAJOR:
-            _canonical_json(_canonical_value(getattr(spec, field)))
-
-    rendered = spec.rendered_spec()
-    fields = {}
-    for field, field_info in type(rendered).model_fields.items():
-        if field in {"name", "namespace", "node_type"}:
-            continue
-        if field_info.exclude is True:
-            continue
-        if type(rendered).field_change_tier(field) != ChangeTier.MAJOR:
-            continue
-        fields[field] = _canonical_field_value(
-            rendered,
-            field,
-            resolved_columns=resolved_columns,
-            preserve_order=(
-                type(rendered).field_order_change_tier(field) == ChangeTier.MAJOR
-            ),
-        )
-    return {
-        "domain": "datajunction/node-semantic",
-        "node_type": _canonical_value(rendered.node_type),
-        "fields": fields,
-    }
-
-
-def _semantic_fingerprint_v1(
-    spec: NodeSpec,
-    parent_fingerprints: Iterable[SemanticFingerprint],
-    *,
-    resolved_columns: list[ColumnSpec] | None,
-) -> SemanticFingerprint:
-    node_payload = _semantic_fingerprint_payload(
-        spec,
-        resolved_columns=resolved_columns,
-    )
-    node_digest = hashlib.sha256(
-        _canonical_json(node_payload).encode("utf-8"),
-    ).hexdigest()
-    parents = list(parent_fingerprints)
-    if any(parent.version != 1 for parent in parents):
-        raise ValueError("Parent fingerprint version does not match node version")
-    payload = {
-        "domain": "datajunction/node-semantic-merkle",
-        "version": 1,
-        "node": node_digest,
-        "parents": sorted({parent.digest for parent in parents}),
-    }
-    digest = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
-    return SemanticFingerprint(version=1, digest=digest)
-
-
-_SEMANTIC_FINGERPRINT_BUILDERS: dict[
-    int,
-    Callable[..., SemanticFingerprint],
-] = {
-    1: _semantic_fingerprint_v1,
-}
-
-
-def _canonical_node_diff(
-    one: NodeSpec,
-    two: NodeSpec,
-    *,
-    resolved_columns: list[ColumnSpec] | None,
-    other_resolved_columns: list[ColumnSpec] | None,
-) -> tuple[list[str], list[str]]:
-    if one.node_type != two.node_type:
-        return ["node_type"], []
-
-    rendered_one = one.rendered_spec()
-    rendered_two = two.rendered_spec()
-
-    changed_fields = []
-    reordered_fields = []
-    for field, field_info in type(rendered_two).model_fields.items():
-        if field in {"name", "namespace", "node_type"}:
-            continue
-        if isinstance(rendered_two, MetricSpec) and field == "unit_structured":
-            continue
-        if field_info.exclude is True and field != "unit_enum":
-            continue
-        if type(rendered_two).field_change_tier(field) == ChangeTier.NONE:
-            continue
-        if field == "display_name" and getattr(rendered_two, field) is None:
-            continue
-
-        left = _canonical_field_value(
-            rendered_one,
-            field,
-            resolved_columns=resolved_columns,
-        )
-        right = _canonical_field_value(
-            rendered_two,
-            field,
-            resolved_columns=other_resolved_columns,
-        )
-        if left != right:
-            changed_fields.append(field)
-            continue
-
-        if type(rendered_two).field_order_change_tier(field) == ChangeTier.NONE:
-            continue
-        left_ordered = _canonical_field_value(
-            rendered_one,
-            field,
-            resolved_columns=resolved_columns,
-            preserve_order=True,
-        )
-        right_ordered = _canonical_field_value(
-            rendered_two,
-            field,
-            resolved_columns=other_resolved_columns,
-            preserve_order=True,
-        )
-        if left_ordered != right_ordered:
-            reordered_fields.append(field)
-
-    return changed_fields, reordered_fields
-
-
 def _norm(v: Any) -> Any:
     """Normalize falsy string/None to None so that '' and None compare equal."""
     if isinstance(v, str):
@@ -2061,6 +1735,8 @@ def eq_columns(
       - If a column is missing display_name or description, it's treated as empty string.
     If the compare_types flag is False, the column types will not be compared.
     """
+    from datajunction_server.semantic_fingerprints.canonical import normalized_column
+
     a_map = {col.name: col for col in a or []}
     b_map = {col.name: col for col in b or []}
     # For source nodes (compare_types=True), column additions and removals from
@@ -2070,13 +1746,13 @@ def eq_columns(
         return False
     a_cols, b_cols = [], []
     for col_name in sorted(set(a_map).union(b_map)):
-        a_col = _normalized_column(
+        a_col = normalized_column(
             a_map.get(col_name),
             col_name,
             b_map[col_name].type if col_name in b_map else "",
             compare_types,
         )
-        b_col = _normalized_column(
+        b_col = normalized_column(
             b_map.get(col_name),
             col_name,
             a_map[col_name].type if col_name in a_map else "",
diff --git a/datajunction-server/datajunction_server/models/semantic_fingerprint.py b/datajunction-server/datajunction_server/models/semantic_fingerprint.py
new file mode 100644
index 000000000..aba526481
--- /dev/null
+++ b/datajunction-server/datajunction_server/models/semantic_fingerprint.py
@@ -0,0 +1,27 @@
+"""Models and version constants for semantic fingerprints."""
+
+from pydantic import BaseModel, Field, field_validator
+
+
+LATEST_SEMANTIC_FINGERPRINT_VERSION = 1
+SUPPORTED_SEMANTIC_FINGERPRINT_VERSIONS = frozenset(
+    {LATEST_SEMANTIC_FINGERPRINT_VERSION},
+)
+
+
+class SemanticFingerprint(BaseModel):
+    """A versioned digest of a node's semantic definition."""
+
+    version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION
+    digest: str = Field(
+        min_length=64,
+        max_length=64,
+        pattern=r"^[0-9a-f]+$",
+    )
+
+    @field_validator("version")
+    @classmethod
+    def validate_version(cls, version: int) -> int:
+        if version not in SUPPORTED_SEMANTIC_FINGERPRINT_VERSIONS:
+            raise ValueError(f"Unsupported semantic fingerprint version: {version}")
+        return version
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
new file mode 100644
index 000000000..01fa6d2f5
--- /dev/null
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
@@ -0,0 +1,8 @@
+"""Semantic fingerprint construction and comparison."""
+
+from datajunction_server.semantic_fingerprints.engine import (
+    canonical_diff,
+    fingerprint_node,
+)
+
+__all__ = ["canonical_diff", "fingerprint_node"]
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/canonical.py b/datajunction-server/datajunction_server/semantic_fingerprints/canonical.py
new file mode 100644
index 000000000..b81167565
--- /dev/null
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/canonical.py
@@ -0,0 +1,270 @@
+"""Canonical node values shared by comparison and fingerprinting."""
+
+import json
+import math
+from collections.abc import Iterable
+from decimal import Decimal
+from enum import Enum
+from typing import Any, get_args, get_origin
+
+from pydantic import BaseModel
+
+from datajunction_server.models.base import labelize
+from datajunction_server.models.deployment import (
+    ChangeTier,
+    ColumnSpec,
+    CubeSpec,
+    DimensionJoinLinkSpec,
+    DimensionReferenceLinkSpec,
+    LinkableNodeSpec,
+    MetricSpec,
+    NodeSpec,
+    SourceSpec,
+)
+from datajunction_server.models.node import MetricDirection
+
+
+def canonical_json(value: Any) -> str:
+    """Serialize a fingerprint value deterministically."""
+    return json.dumps(
+        value,
+        allow_nan=False,
+        ensure_ascii=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
+def canonical_value(value: Any) -> Any:
+    """Convert supported values to deterministic JSON-compatible values."""
+    if isinstance(value, Enum):
+        return canonical_value(value.value)
+    if isinstance(value, BaseModel):
+        return canonical_value(value.model_dump(mode="python"))
+    if isinstance(value, dict):
+        if any(not isinstance(key, str) for key in value):
+            raise TypeError("Semantic fingerprint mappings require string keys")
+        return {key: canonical_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [canonical_value(item) for item in value]
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise ValueError("Semantic fingerprint values must be finite")
+        if value == 0 or value.is_integer():
+            return int(value)
+        return value
+    if isinstance(value, Decimal):
+        if not value.is_finite():
+            raise ValueError("Semantic fingerprint values must be finite")
+        if value == value.to_integral_value():
+            return int(value)
+        return {"decimal": format(value.normalize(), "f")}
+    if value is None or isinstance(value, (str, int)):
+        return value
+    raise TypeError(
+        f"Unsupported semantic fingerprint value: {type(value).__name__}",
+    )
+
+
+def canonical_sequence(
+    values: Iterable[Any],
+    *,
+    preserve_order: bool,
+) -> list[Any]:
+    """Canonicalize a sequence while removing duplicate semantic values."""
+    unique: dict[str, Any] = {}
+    for value in values:
+        canonical = canonical_value(value)
+        unique.setdefault(canonical_json(canonical), canonical)
+    return (
+        list(unique.values())
+        if preserve_order
+        else [unique[key] for key in sorted(unique)]
+    )
+
+
+def annotation_contains_list(annotation: Any) -> bool:
+    """Return whether an annotation contains a list type."""
+    return get_origin(annotation) is list or any(
+        annotation_contains_list(argument) for argument in get_args(annotation)
+    )
+
+
+def normalized_column(
+    column: ColumnSpec | None,
+    name: str,
+    fallback_type: str | None,
+    compare_types: bool,
+) -> ColumnSpec:
+    normalized = (
+        column.model_copy()
+        if column
+        else ColumnSpec(
+            name=name,
+            display_name=labelize(name),
+            type=fallback_type or "",
+            attributes=[],
+        )
+    )
+    normalized.display_name = normalized.display_name or labelize(name)
+    normalized.description = normalized.description or ""
+    normalized.attributes = sorted(set(normalized.attributes) - {"primary_key"})
+    if not compare_types:
+        normalized.type = ""
+    return normalized
+
+
+def canonical_columns(
+    columns: list[ColumnSpec] | None,
+    *,
+    compare_types: bool,
+) -> list[Any]:
+    """Canonicalize authored and resolved node columns."""
+    column_map = {column.name: column for column in columns or []}
+    canonical = []
+    for name in sorted(column_map):
+        column = column_map[name]
+        normalized = normalized_column(column, name, column.type, compare_types)
+        if not compare_types:
+            default = normalized_column(None, name, column.type, compare_types)
+            if normalized == default:
+                continue
+        canonical.append(canonical_value(normalized))
+    return canonical
+
+
+def canonical_dimension_links(
+    links: list[DimensionJoinLinkSpec | DimensionReferenceLinkSpec] | None,
+    *,
+    preserve_order: bool,
+) -> list[Any]:
+    """Canonicalize dimension links by their semantic comparison key."""
+    return canonical_sequence(
+        (link._comparison_key() for link in links or []),
+        preserve_order=preserve_order,
+    )
+
+
+def canonical_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]:
+    """Canonicalize the authored partition configuration of cube columns."""
+    return {
+        column.name: canonical_value(column.partition)
+        for column in columns or []
+        if column.partition
+    }
+
+
+def canonical_field_value(
+    spec: NodeSpec,
+    field: str,
+    *,
+    resolved_columns: list[ColumnSpec] | None = None,
+    preserve_order: bool = False,
+    structural_version: int | None = None,
+) -> Any:
+    """Return the canonical semantic value of one node field."""
+    value = getattr(spec, field)
+    if field == "query":
+        from datajunction_server.sql.parsing.structural import serialize_ast
+
+        return (
+            serialize_ast(spec.query_ast, version=structural_version)
+            if spec.query_ast is not None
+            else spec.rendered_query
+        )
+    if field == "columns":
+        if isinstance(spec, CubeSpec):
+            return canonical_cube_columns(spec.rendered_columns)
+        return canonical_columns(
+            resolved_columns
+            if isinstance(spec, SourceSpec) and resolved_columns is not None
+            else value,
+            compare_types=isinstance(spec, SourceSpec),
+        )
+    if field == "dimension_links" and isinstance(spec, LinkableNodeSpec):
+        return canonical_dimension_links(
+            spec.dimension_links,
+            preserve_order=preserve_order,
+        )
+    if field == "unit_enum" and isinstance(spec, MetricSpec):
+        return canonical_value(spec._canonical_unit())
+    if field == "direction" and isinstance(spec, MetricSpec):
+        value = value or MetricDirection.NEUTRAL
+    if field == "description":
+        value = value or None
+    if field == "custom_metadata":
+        value = value or {}
+    if value is None and annotation_contains_list(
+        type(spec).model_fields[field].annotation,
+    ):
+        value = []
+
+    canonical = canonical_value(value)
+    return (
+        canonical_sequence(canonical, preserve_order=preserve_order)
+        if isinstance(canonical, list)
+        else canonical
+    )
+
+
+def canonical_node_diff(
+    one: NodeSpec,
+    two: NodeSpec,
+    *,
+    resolved_columns: list[ColumnSpec] | None,
+    other_resolved_columns: list[ColumnSpec] | None,
+) -> tuple[list[str], list[str]]:
+    """Compare two specs using their canonical semantic values."""
+    if one.node_type != two.node_type:
+        return ["node_type"], []
+
+    rendered_one = one.rendered_spec()
+    rendered_two = two.rendered_spec()
+    changed_fields = []
+    reordered_fields = []
+    for field, field_info in type(rendered_two).model_fields.items():
+        if field in {"name", "namespace", "node_type"}:
+            continue
+        if isinstance(rendered_two, MetricSpec) and field == "unit_structured":
+            continue
+        if field_info.exclude is True and field != "unit_enum":
+            continue
+        if type(rendered_two).field_change_tier(field) == ChangeTier.NONE:
+            continue
+        if field == "display_name" and getattr(rendered_two, field) is None:
+            continue
+
+        left = canonical_field_value(
+            rendered_one,
+            field,
+            resolved_columns=resolved_columns,
+        )
+        right = canonical_field_value(
+            rendered_two,
+            field,
+            resolved_columns=other_resolved_columns,
+        )
+        if left != right:
+            changed_fields.append(field)
+            continue
+
+        if type(rendered_two).field_order_change_tier(field) == ChangeTier.NONE:
+            continue
+        left_ordered = canonical_field_value(
+            rendered_one,
+            field,
+            resolved_columns=resolved_columns,
+            preserve_order=True,
+        )
+        right_ordered = canonical_field_value(
+            rendered_two,
+            field,
+            resolved_columns=other_resolved_columns,
+            preserve_order=True,
+        )
+        if left_ordered != right_ordered:
+            reordered_fields.append(field)
+
+    return changed_fields, reordered_fields
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/engine.py b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
new file mode 100644
index 000000000..ce7f14732
--- /dev/null
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
@@ -0,0 +1,55 @@
+"""Version dispatch for semantic fingerprints."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from typing import TYPE_CHECKING
+
+from datajunction_server.models.semantic_fingerprint import (
+    LATEST_SEMANTIC_FINGERPRINT_VERSION,
+    SemanticFingerprint,
+)
+from datajunction_server.semantic_fingerprints.canonical import canonical_node_diff
+from datajunction_server.semantic_fingerprints.v1 import build_fingerprint
+
+if TYPE_CHECKING:
+    from datajunction_server.models.deployment import ColumnSpec, NodeSpec
+
+
+_BUILDERS: dict[int, Callable[..., SemanticFingerprint]] = {
+    1: build_fingerprint,
+}
+
+
+def fingerprint_node(
+    spec: NodeSpec,
+    version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION,
+    *,
+    parent_fingerprints: Iterable[SemanticFingerprint] = (),
+    resolved_columns: list[ColumnSpec] | None = None,
+) -> SemanticFingerprint:
+    """Return a fingerprint of a node's semantic definition."""
+    builder = _BUILDERS.get(version)
+    if builder is None:
+        raise ValueError(f"Unsupported semantic fingerprint version: {version}")
+    return builder(
+        spec,
+        parent_fingerprints,
+        resolved_columns=resolved_columns,
+    )
+
+
+def canonical_diff(
+    one: NodeSpec,
+    two: NodeSpec,
+    *,
+    resolved_columns: list[ColumnSpec] | None = None,
+    other_resolved_columns: list[ColumnSpec] | None = None,
+) -> tuple[list[str], list[str]]:
+    """Compare two specs using the same canonical values as fingerprints."""
+    return canonical_node_diff(
+        one,
+        two,
+        resolved_columns=resolved_columns,
+        other_resolved_columns=other_resolved_columns,
+    )
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
new file mode 100644
index 000000000..557ae43de
--- /dev/null
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
@@ -0,0 +1,106 @@
+"""Frozen semantic fingerprint version 1."""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Iterable
+from typing import TYPE_CHECKING
+
+from datajunction_server.models.deployment import (
+    ChangeTier,
+    CubeSpec,
+    DimensionSpec,
+    MetricSpec,
+    SourceSpec,
+    TransformSpec,
+)
+from datajunction_server.models.semantic_fingerprint import SemanticFingerprint
+from datajunction_server.semantic_fingerprints.canonical import (
+    canonical_field_value,
+    canonical_json,
+    canonical_value,
+)
+
+if TYPE_CHECKING:
+    from datajunction_server.models.deployment import ColumnSpec, NodeSpec
+
+
+_FIELDS_BY_SPEC_TYPE: dict[type[NodeSpec], tuple[str, ...]] = {
+    SourceSpec: (
+        "columns",
+        "dimension_links",
+        "primary_key",
+        "catalog",
+        "schema_",
+        "table",
+    ),
+    TransformSpec: (
+        "columns",
+        "dimension_links",
+        "primary_key",
+        "query",
+    ),
+    DimensionSpec: (
+        "columns",
+        "dimension_links",
+        "primary_key",
+        "query",
+    ),
+    MetricSpec: ("query", "required_dimensions"),
+    CubeSpec: ("metrics", "dimensions", "filters", "columns"),
+}
+
+
+def semantic_fields(spec_type: type[NodeSpec]) -> tuple[str, ...]:
+    """Return the frozen field projection for a concrete node type."""
+    try:
+        return _FIELDS_BY_SPEC_TYPE[spec_type]
+    except KeyError as exc:
+        raise TypeError(
+            f"No semantic fingerprint fields for {spec_type.__name__}",
+        ) from exc
+
+
+def build_fingerprint(
+    spec: NodeSpec,
+    parent_fingerprints: Iterable[SemanticFingerprint],
+    *,
+    resolved_columns: list[ColumnSpec] | None,
+) -> SemanticFingerprint:
+    """Build a version 1 semantic fingerprint."""
+    fingerprint_fields = semantic_fields(type(spec))
+    for field in fingerprint_fields:
+        canonical_json(canonical_value(getattr(spec, field)))
+
+    rendered = spec.rendered_spec()
+    fields = {
+        field: canonical_field_value(
+            rendered,
+            field,
+            resolved_columns=resolved_columns,
+            preserve_order=(
+                type(rendered).field_order_change_tier(field) == ChangeTier.MAJOR
+            ),
+            structural_version=1,
+        )
+        for field in fingerprint_fields
+    }
+    node_payload = {
+        "domain": "datajunction/node-semantic",
+        "node_type": canonical_value(rendered.node_type),
+        "fields": fields,
+    }
+    node_digest = hashlib.sha256(
+        canonical_json(node_payload).encode("utf-8"),
+    ).hexdigest()
+    parents = list(parent_fingerprints)
+    if any(parent.version != 1 for parent in parents):
+        raise ValueError("Parent fingerprint version does not match node version")
+    payload = {
+        "domain": "datajunction/node-semantic-merkle",
+        "version": 1,
+        "node": node_digest,
+        "parents": sorted({parent.digest for parent in parents}),
+    }
+    digest = hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest()
+    return SemanticFingerprint(version=1, digest=digest)
diff --git a/datajunction-server/datajunction_server/sql/parsing/structural.py b/datajunction-server/datajunction_server/sql/parsing/structural.py
new file mode 100644
index 000000000..cdec4986d
--- /dev/null
+++ b/datajunction-server/datajunction_server/sql/parsing/structural.py
@@ -0,0 +1,137 @@
+"""Versioned structural serialization for parsed SQL."""
+
+from collections.abc import Callable
+from decimal import Decimal
+from enum import Enum
+import math
+from typing import Any
+
+from datajunction_server.sql.parsing.ast import Node
+from datajunction_server.sql.parsing.types import ColumnType
+
+
+_AST_NODE_TAGS_V1 = frozenset(
+    {
+        "Alias",
+        "ArithmeticUnaryOp",
+        "Between",
+        "BinaryOp",
+        "Boolean",
+        "Case",
+        "Cast",
+        "Column",
+        "DefaultName",
+        "Frame",
+        "FrameBound",
+        "From",
+        "Function",
+        "FunctionTable",
+        "FunctionTableExpression",
+        "Hint",
+        "In",
+        "InlineTable",
+        "Interval",
+        "IntervalUnit",
+        "IsBoolean",
+        "IsDistinctFrom",
+        "IsNull",
+        "Join",
+        "JoinCriteria",
+        "Lambda",
+        "LateralView",
+        "Like",
+        "Name",
+        "Null",
+        "Number",
+        "Organization",
+        "Over",
+        "Query",
+        "QueryParameter",
+        "Relation",
+        "Rlike",
+        "Select",
+        "SelectExpression",
+        "SetOp",
+        "SortItem",
+        "String",
+        "Struct",
+        "Subscript",
+        "Table",
+        "UnaryOp",
+        "UnNamed",
+        "Wildcard",
+    },
+)
+
+
+def _serialize_number_v1(value: int | float | Decimal) -> int | float | dict[str, str]:
+    if isinstance(value, bool):
+        raise TypeError("Boolean values are not SQL numbers")
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise ValueError("Structural SQL numbers must be finite")
+        if value == 0 or value.is_integer():
+            return int(value)
+        return value
+    if isinstance(value, Decimal):
+        if not value.is_finite():
+            raise ValueError("Structural SQL numbers must be finite")
+        if value == value.to_integral_value():
+            return int(value)
+        return {"decimal": format(value.normalize(), "f")}
+    return value
+
+
+def _serialize_ast_v1(query_ast: Node) -> Any:
+    """Serialize an AST without SQL rendering or Python module-qualified names."""
+
+    def serialize(value: Any) -> Any:
+        if isinstance(value, Node):
+            tag = type(value).__name__
+            if tag not in _AST_NODE_TAGS_V1:
+                raise TypeError(f"Unsupported structural SQL node: {tag}")
+            return {
+                "type": tag,
+                "fields": {
+                    name: serialize(field_value)
+                    for name, field_value in value.fields(
+                        flat=False,
+                        nodes_only=False,
+                        obfuscated=False,
+                        nones=True,
+                        named=True,
+                    )
+                },
+            }
+        if isinstance(value, ColumnType):
+            return {"type": "column_type", "value": str(value)}
+        if isinstance(value, Enum):
+            return serialize(value.value)
+        if isinstance(value, Decimal):
+            return _serialize_number_v1(value)
+        if isinstance(value, float):
+            return _serialize_number_v1(value)
+        if isinstance(value, (list, tuple)):
+            return [serialize(item) for item in value]
+        if value is None or isinstance(value, (str, int, bool)):
+            return value
+        raise TypeError(f"Unsupported structural SQL value: {type(value).__name__}")
+
+    return serialize(query_ast)
+
+
+_LATEST_VERSION = 1
+_SERIALIZERS: dict[int, Callable[[Node], Any]] = {
+    1: _serialize_ast_v1,
+}
+
+
+def serialize_ast(query_ast: Node, *, version: int | None = None) -> Any:
+    """Serialize an AST using a stable structural format."""
+    selected_version = _LATEST_VERSION if version is None else version
+    serializer = _SERIALIZERS.get(selected_version)
+    if serializer is None:
+        raise ValueError(
+            f"Unsupported structural SQL serialization version: {selected_version}",
+        )
+    return serializer(query_ast)
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index 5b0fd4dfa..5b7644edf 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -3,7 +3,6 @@
 import subprocess
 import sys
 from datetime import date
-from typing import Any, ClassVar
 
 import pytest
 from pydantic import ValidationError
@@ -39,6 +38,12 @@
     eq_or_fallback,
     fold_change_tiers,
 )
+from datajunction_server.semantic_fingerprints.canonical import (
+    canonical_json,
+    canonical_sequence,
+    canonical_value,
+)
+from datajunction_server.semantic_fingerprints.v1 import semantic_fields
 from datajunction_server.models.materialization import (
     CoverageSpec,
     MaterializationStrategy,
@@ -1694,9 +1699,9 @@ def fingerprint(spec: NodeSpec) -> SemanticFingerprint:
 
 GOLDEN_FINGERPRINTS = {
     "source": "71dcbc388988c2bdd850670427710384687b58565ee38ca392dc220adfed868d",
-    "transform": "797af072eb15831df786149716171f0a400af7f35f6031f8eee3939733cab9c4",
-    "dimension": "965282bbe90a4fcc66576fba3550df42c09ff4dc51be2ee433a3897d48026edb",
-    "metric": "738703ffa80a72b00fc829697cf259aac317dc34290cb1aae4c0678a34f3a948",
+    "transform": "978e692880c7bcfb1bd78ece85895a1ec1558e85377b064a8dac3f3719cff2a5",
+    "dimension": "f7b3c87a61fdadf9997432fd9334befdf43f2488874ef555e3f7d4c4ba86e3e1",
+    "metric": "a3fde7af5dbd00d194805af33fc213bca52244c7cdedf1f7363ec52d2f6d4116",
     "cube": "9b0a56d974d1e3769bc2db94e2cfbae7a6a4839f664eebd2a4387ef112ceea81",
 }
 
@@ -1747,39 +1752,50 @@ def test_semantic_fingerprint_normalizes_empty_and_resolved_source_columns():
     ) == fingerprint(CubeSpec(name="cube", metrics=[], dimensions=[], filters=[]))
 
 
-def test_semantic_fingerprint_mapping_order_is_stable():
-    class CanonicalSpec(NodeSpec):
-        semantic_mapping: dict[str, Any]
-        values: list[int]
-        FIELD_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
-            "semantic_mapping": ChangeTier.MAJOR,
-            "values": ChangeTier.MAJOR,
-        }
-        FIELD_ORDER_CHANGE_TIERS: ClassVar[dict[str, ChangeTier]] = {
-            "values": ChangeTier.MAJOR,
-        }
-
-    common = {"name": "mapped", "node_type": NodeType.SOURCE, "values": [1, 2]}
-    first = CanonicalSpec(
-        **common,
-        semantic_mapping={"outer": {"a": 1, "b": 2}, "value": 3},
+def test_semantic_fingerprint_canonical_values_are_stable():
+    first = {"outer": {"a": 1, "b": 2}, "value": 3}
+    second = {"value": 3, "outer": {"b": 2, "a": 1}}
+    assert canonical_json(canonical_value(first)) == canonical_json(
+        canonical_value(second),
     )
-    second = CanonicalSpec(
-        **common,
-        semantic_mapping={"value": 3, "outer": {"b": 2, "a": 1}},
-    )
-    assert fingerprint(first) == fingerprint(second)
     with pytest.raises(TypeError, match="string keys"):
-        fingerprint(first.model_copy(update={"semantic_mapping": {1: "value"}}))
-    assert fingerprint(first) != fingerprint(
-        first.model_copy(update={"values": [2, 1]}),
+        canonical_value({1: "value"})
+    assert canonical_sequence(
+        [1, 2],
+        preserve_order=True,
+    ) != canonical_sequence(
+        [2, 1],
+        preserve_order=True,
     )
     with pytest.raises(TypeError, match="Unsupported"):
-        fingerprint(first.model_copy(update={"semantic_mapping": {"bad": object()}}))
+        canonical_value({"bad": object()})
     with pytest.raises(ValueError, match="must be finite"):
-        fingerprint(
-            first.model_copy(update={"semantic_mapping": {"bad": float("nan")}}),
-        )
+        canonical_value({"bad": float("nan")})
+
+
+def test_semantic_fingerprint_normalizes_equivalent_numbers():
+    assert canonical_value({"value": 1}) == canonical_value({"value": 1.0})
+    assert canonical_value({"value": -0.0}) == canonical_value({"value": 0})
+
+    sql_integer = TransformSpec(name="sql_number", query="SELECT 1")
+    sql_integral_float = TransformSpec(name="sql_number", query="SELECT 1.0")
+    assert sql_integer.canonical_diff(sql_integral_float) == ([], [])
+    assert fingerprint(sql_integer) == fingerprint(sql_integral_float)
+
+
+@pytest.mark.parametrize(
+    "spec_type",
+    [SourceSpec, TransformSpec, DimensionSpec, MetricSpec, CubeSpec],
+)
+def test_semantic_fingerprint_v1_projection_is_explicit(spec_type):
+    current_major_fields = {
+        field
+        for field, field_info in spec_type.model_fields.items()
+        if field not in {"name", "namespace", "node_type"}
+        and field_info.exclude is not True
+        and spec_type.field_change_tier(field) == ChangeTier.MAJOR
+    }
+    assert set(semantic_fields(spec_type)) == current_major_fields
 
 
 def test_semantic_fingerprint_renders_prefixes_and_canonicalizes_sql():
@@ -2044,9 +2060,13 @@ def test_semantic_fingerprint_normalizes_primary_keys_and_cube_ordering():
         query="SELECT 1",
         required_dimensions=["one", "two"],
     )
-    assert fingerprint(metric) == fingerprint(
-        metric.model_copy(update={"required_dimensions": ["two", "one"]}),
+    reordered_metric = metric.model_copy(
+        update={"required_dimensions": ["two", "one", "one"]},
     )
+    assert metric == reordered_metric
+    assert metric.canonical_diff(reordered_metric) == ([], [])
+    assert fingerprint(metric) == fingerprint(reordered_metric)
+    assert MetricSpec.field_change_tier("columns") == ChangeTier.NONE
     assert fingerprint(cube) == fingerprint(
         cube.model_copy(
             update={"metrics": [*cube.metrics, "${prefix}order_count"]},
diff --git a/datajunction-server/tests/sql/parsing/test_structural.py b/datajunction-server/tests/sql/parsing/test_structural.py
new file mode 100644
index 000000000..723d5ba5d
--- /dev/null
+++ b/datajunction-server/tests/sql/parsing/test_structural.py
@@ -0,0 +1,32 @@
+from dataclasses import dataclass
+
+import pytest
+
+from datajunction_server.sql.parsing import ast
+from datajunction_server.sql.parsing.backends.antlr4 import parse
+from datajunction_server.sql.parsing.structural import serialize_ast
+
+
+def test_serialize_ast_is_structural_and_numeric():
+    integer = serialize_ast(parse("SELECT 1"))
+    integral_float = serialize_ast(parse("SELECT 1.0"))
+    typed = serialize_ast(
+        parse("SELECT CAST(value AS DECIMAL(10, 2)) FROM source"),
+    )
+
+    assert integer == integral_float
+    assert integer["type"] == "Query"
+    assert "datajunction_server." not in str(integer)
+    assert typed["type"] == "Query"
+
+
+def test_serialize_ast_rejects_unclassified_nodes_and_versions():
+    @dataclass(eq=False)
+    class FutureNode(ast.Node):
+        def __str__(self) -> str:
+            return "future"
+
+    with pytest.raises(TypeError, match="Unsupported structural SQL node: FutureNode"):
+        serialize_ast(FutureNode())
+    with pytest.raises(ValueError, match="Unsupported structural SQL serialization"):
+        serialize_ast(parse("SELECT 1"), version=2)

From 8dfb7d7b5d7fc5db01cbd7a73b627b8921313e89 Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 15:46:48 -0700
Subject: [PATCH 06/11] Use semantic normalization terminology

Reserve canonical for final JSON serialization and name comparison helpers by behavior so the fingerprint API is easier to read.
---
 .../datajunction_server/models/deployment.py  | 50 ++++++------
 .../semantic_fingerprints/__init__.py         |  4 +-
 .../semantic_fingerprints/engine.py           | 10 ++-
 .../{canonical.py => normalization.py}        | 81 ++++++++++---------
 .../semantic_fingerprints/v1.py               | 12 +--
 .../tests/models/deployment_test.py           | 46 +++++------
 6 files changed, 103 insertions(+), 100 deletions(-)
 rename datajunction-server/datajunction_server/semantic_fingerprints/{canonical.py => normalization.py} (77%)

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index 1564bf374..031f62bc7 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -718,16 +718,16 @@ def semantic_fingerprint(
             resolved_columns=resolved_columns,
         )
 
-    def canonical_diff(
+    def semantic_diff(
         self,
         other: "NodeSpec",
         *,
         resolved_columns: list[ColumnSpec] | None = None,
         other_resolved_columns: list[ColumnSpec] | None = None,
     ) -> tuple[list[str], list[str]]:
-        """Compare two specs using the same canonical values as fingerprints."""
+        """Compare two specs using the same normalized values as fingerprints."""
         from datajunction_server.semantic_fingerprints import (
-            canonical_diff as compare_semantics,
+            semantic_diff as compare_semantics,
         )
 
         return compare_semantics(
@@ -806,7 +806,7 @@ def unclassified_fields(cls) -> list[str]:
     @classmethod
     def unclassified_list_order_fields(cls) -> list[str]:
         """List fields without an explicit order classification."""
-        from datajunction_server.semantic_fingerprints.canonical import (
+        from datajunction_server.semantic_fingerprints.normalization import (
             annotation_contains_list,
         )
 
@@ -913,8 +913,8 @@ def links_mapping(self) -> dict[tuple[str, str | None], DimensionLinkSpec]:
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, LinkableNodeSpec):
             return False  # pragma: no cover
-        from datajunction_server.semantic_fingerprints.canonical import (
-            canonical_dimension_links,
+        from datajunction_server.semantic_fingerprints.normalization import (
+            normalize_dimension_links,
         )
 
         return (
@@ -924,11 +924,11 @@ def __eq__(self, other: object) -> bool:
                 other.columns,
                 compare_types=self.node_type == NodeType.SOURCE,
             )
-            and canonical_dimension_links(
+            and normalize_dimension_links(
                 self.dimension_links,
                 preserve_order=False,
             )
-            == canonical_dimension_links(
+            == normalize_dimension_links(
                 other.dimension_links,
                 preserve_order=False,
             )
@@ -1094,7 +1094,7 @@ def __init__(self, **data: Any):
     @property
     def unit(self) -> str | dict | None:
         """
-        Return the canonical metric unit value for serialization.
+        Return the normalized metric unit value for serialization.
 
         Returns:
           - `None` if no unit is set.
@@ -1106,7 +1106,7 @@ def unit(self) -> str | dict | None:
         shape should read `column.unit` on the metric's output column.
         """
         if self.unit_structured is not None:
-            # Canonical dict shape (JSON-friendly, no None values).
+            # Normalized dict shape (JSON-friendly, no None values).
             return unit_to_dict(self.unit_structured)
         if self.unit_enum is None or self.unit_enum == MetricUnit.UNKNOWN:
             return None
@@ -1133,14 +1133,14 @@ def model_dump(self, **kwargs):  # pragma: no cover
         base["unit"] = self.unit
         return base
 
-    def _canonical_unit(self) -> "Unit | None":
+    def _normalized_unit(self) -> "Unit | None":
         """
-        Reduce both legacy and structured inputs to the same canonical Unit
+        Reduce both legacy and structured inputs to the same normalized Unit
         instance for equality comparisons. Returns None when the metric has
         no unit (or only the UNKNOWN sentinel). Two specs that author the
         same conceptual unit via different input shapes (`unit: dollar` vs
         `unit: {kind: currency, code: USD}`) produce equal frozen Unit
-        instances — so __eq__ doesn't falsely report drift between YAML and
+        instances, so __eq__ doesn't falsely report drift between YAML and
         DB-roundtripped specs.
         """
         if self.unit_structured is not None:
@@ -1152,23 +1152,23 @@ def _canonical_unit(self) -> "Unit | None":
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, MetricSpec):
             return False
-        from datajunction_server.semantic_fingerprints.canonical import (
-            canonical_sequence,
+        from datajunction_server.semantic_fingerprints.normalization import (
+            normalize_sequence,
         )
 
         return (
             super().__eq__(other)
             and self.query_ast.compare(other.query_ast)
-            and canonical_sequence(
+            and normalize_sequence(
                 self.rendered_required_dimensions,
                 preserve_order=False,
             )
-            == canonical_sequence(
+            == normalize_sequence(
                 other.rendered_required_dimensions,
                 preserve_order=False,
             )
             and eq_or_fallback(self.direction, other.direction, MetricDirection.NEUTRAL)
-            and self._canonical_unit() == other._canonical_unit()
+            and self._normalized_unit() == other._normalized_unit()
             and self.significant_digits == other.significant_digits
             and self.min_decimal_exponent == other.min_decimal_exponent
             and self.max_decimal_exponent == other.max_decimal_exponent
@@ -1368,13 +1368,13 @@ def __eq__(self, other: object) -> bool:
 
         # Compare only partition config for user-specified columns.
         # Cube element columns (types, order, attributes) are auto-derived and ignored.
-        from datajunction_server.semantic_fingerprints.canonical import (
-            canonical_cube_columns,
+        from datajunction_server.semantic_fingerprints.normalization import (
+            normalize_cube_columns,
         )
 
-        return canonical_cube_columns(
+        return normalize_cube_columns(
             self.rendered_columns,
-        ) == canonical_cube_columns(other.rendered_columns)
+        ) == normalize_cube_columns(other.rendered_columns)
 
 
 NodeUnion = Annotated[
@@ -1735,7 +1735,7 @@ def eq_columns(
       - If a column is missing display_name or description, it's treated as empty string.
     If the compare_types flag is False, the column types will not be compared.
     """
-    from datajunction_server.semantic_fingerprints.canonical import normalized_column
+    from datajunction_server.semantic_fingerprints.normalization import normalize_column
 
     a_map = {col.name: col for col in a or []}
     b_map = {col.name: col for col in b or []}
@@ -1746,13 +1746,13 @@ def eq_columns(
         return False
     a_cols, b_cols = [], []
     for col_name in sorted(set(a_map).union(b_map)):
-        a_col = normalized_column(
+        a_col = normalize_column(
             a_map.get(col_name),
             col_name,
             b_map[col_name].type if col_name in b_map else "",
             compare_types,
         )
-        b_col = normalized_column(
+        b_col = normalize_column(
             b_map.get(col_name),
             col_name,
             a_map[col_name].type if col_name in a_map else "",
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
index 01fa6d2f5..368e16814 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
@@ -1,8 +1,8 @@
 """Semantic fingerprint construction and comparison."""
 
 from datajunction_server.semantic_fingerprints.engine import (
-    canonical_diff,
     fingerprint_node,
+    semantic_diff,
 )
 
-__all__ = ["canonical_diff", "fingerprint_node"]
+__all__ = ["fingerprint_node", "semantic_diff"]
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/engine.py b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
index ce7f14732..f4cbc1469 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
@@ -9,7 +9,9 @@
     LATEST_SEMANTIC_FINGERPRINT_VERSION,
     SemanticFingerprint,
 )
-from datajunction_server.semantic_fingerprints.canonical import canonical_node_diff
+from datajunction_server.semantic_fingerprints.normalization import (
+    semantic_diff as compare_semantics,
+)
 from datajunction_server.semantic_fingerprints.v1 import build_fingerprint
 
 if TYPE_CHECKING:
@@ -39,15 +41,15 @@ def fingerprint_node(
     )
 
 
-def canonical_diff(
+def semantic_diff(
     one: NodeSpec,
     two: NodeSpec,
     *,
     resolved_columns: list[ColumnSpec] | None = None,
     other_resolved_columns: list[ColumnSpec] | None = None,
 ) -> tuple[list[str], list[str]]:
-    """Compare two specs using the same canonical values as fingerprints."""
-    return canonical_node_diff(
+    """Compare two specs using the same normalized values as fingerprints."""
+    return compare_semantics(
         one,
         two,
         resolved_columns=resolved_columns,
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/canonical.py b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
similarity index 77%
rename from datajunction-server/datajunction_server/semantic_fingerprints/canonical.py
rename to datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
index b81167565..13e3d1462 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/canonical.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
@@ -1,4 +1,4 @@
-"""Canonical node values shared by comparison and fingerprinting."""
+"""Semantic value normalization shared by comparison and fingerprinting."""
 
 import json
 import math
@@ -35,18 +35,18 @@ def canonical_json(value: Any) -> str:
     )
 
 
-def canonical_value(value: Any) -> Any:
+def normalize_value(value: Any) -> Any:
     """Convert supported values to deterministic JSON-compatible values."""
     if isinstance(value, Enum):
-        return canonical_value(value.value)
+        return normalize_value(value.value)
     if isinstance(value, BaseModel):
-        return canonical_value(value.model_dump(mode="python"))
+        return normalize_value(value.model_dump(mode="python"))
     if isinstance(value, dict):
         if any(not isinstance(key, str) for key in value):
             raise TypeError("Semantic fingerprint mappings require string keys")
-        return {key: canonical_value(item) for key, item in value.items()}
+        return {key: normalize_value(item) for key, item in value.items()}
     if isinstance(value, (list, tuple)):
-        return [canonical_value(item) for item in value]
+        return [normalize_value(item) for item in value]
     if isinstance(value, bool):
         return value
     if isinstance(value, float):
@@ -68,16 +68,16 @@ def canonical_value(value: Any) -> Any:
     )
 
 
-def canonical_sequence(
+def normalize_sequence(
     values: Iterable[Any],
     *,
     preserve_order: bool,
 ) -> list[Any]:
-    """Canonicalize a sequence while removing duplicate semantic values."""
+    """Normalize a sequence while removing duplicate semantic values."""
     unique: dict[str, Any] = {}
     for value in values:
-        canonical = canonical_value(value)
-        unique.setdefault(canonical_json(canonical), canonical)
+        normalized = normalize_value(value)
+        unique.setdefault(canonical_json(normalized), normalized)
     return (
         list(unique.values())
         if preserve_order
@@ -92,12 +92,13 @@ def annotation_contains_list(annotation: Any) -> bool:
     )
 
 
-def normalized_column(
+def normalize_column(
     column: ColumnSpec | None,
     name: str,
     fallback_type: str | None,
     compare_types: bool,
 ) -> ColumnSpec:
+    """Fill equivalent column defaults before comparison or hashing."""
     normalized = (
         column.model_copy()
         if column
@@ -116,47 +117,47 @@ def normalized_column(
     return normalized
 
 
-def canonical_columns(
+def normalize_columns(
     columns: list[ColumnSpec] | None,
     *,
     compare_types: bool,
 ) -> list[Any]:
-    """Canonicalize authored and resolved node columns."""
+    """Normalize authored and resolved node columns."""
     column_map = {column.name: column for column in columns or []}
-    canonical = []
+    normalized_columns = []
     for name in sorted(column_map):
         column = column_map[name]
-        normalized = normalized_column(column, name, column.type, compare_types)
+        normalized = normalize_column(column, name, column.type, compare_types)
         if not compare_types:
-            default = normalized_column(None, name, column.type, compare_types)
+            default = normalize_column(None, name, column.type, compare_types)
             if normalized == default:
                 continue
-        canonical.append(canonical_value(normalized))
-    return canonical
+        normalized_columns.append(normalize_value(normalized))
+    return normalized_columns
 
 
-def canonical_dimension_links(
+def normalize_dimension_links(
     links: list[DimensionJoinLinkSpec | DimensionReferenceLinkSpec] | None,
     *,
     preserve_order: bool,
 ) -> list[Any]:
-    """Canonicalize dimension links by their semantic comparison key."""
-    return canonical_sequence(
+    """Normalize dimension links by their semantic comparison key."""
+    return normalize_sequence(
         (link._comparison_key() for link in links or []),
         preserve_order=preserve_order,
     )
 
 
-def canonical_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]:
-    """Canonicalize the authored partition configuration of cube columns."""
+def normalize_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]:
+    """Normalize the authored partition configuration of cube columns."""
     return {
-        column.name: canonical_value(column.partition)
+        column.name: normalize_value(column.partition)
         for column in columns or []
         if column.partition
     }
 
 
-def canonical_field_value(
+def normalize_field(
     spec: NodeSpec,
     field: str,
     *,
@@ -164,7 +165,7 @@ def canonical_field_value(
     preserve_order: bool = False,
     structural_version: int | None = None,
 ) -> Any:
-    """Return the canonical semantic value of one node field."""
+    """Return the normalized semantic value of one node field."""
     value = getattr(spec, field)
     if field == "query":
         from datajunction_server.sql.parsing.structural import serialize_ast
@@ -176,20 +177,20 @@ def canonical_field_value(
         )
     if field == "columns":
         if isinstance(spec, CubeSpec):
-            return canonical_cube_columns(spec.rendered_columns)
-        return canonical_columns(
+            return normalize_cube_columns(spec.rendered_columns)
+        return normalize_columns(
             resolved_columns
             if isinstance(spec, SourceSpec) and resolved_columns is not None
             else value,
             compare_types=isinstance(spec, SourceSpec),
         )
     if field == "dimension_links" and isinstance(spec, LinkableNodeSpec):
-        return canonical_dimension_links(
+        return normalize_dimension_links(
             spec.dimension_links,
             preserve_order=preserve_order,
         )
     if field == "unit_enum" and isinstance(spec, MetricSpec):
-        return canonical_value(spec._canonical_unit())
+        return normalize_value(spec._normalized_unit())
     if field == "direction" and isinstance(spec, MetricSpec):
         value = value or MetricDirection.NEUTRAL
     if field == "description":
@@ -201,22 +202,22 @@ def canonical_field_value(
     ):
         value = []
 
-    canonical = canonical_value(value)
+    normalized = normalize_value(value)
     return (
-        canonical_sequence(canonical, preserve_order=preserve_order)
-        if isinstance(canonical, list)
-        else canonical
+        normalize_sequence(normalized, preserve_order=preserve_order)
+        if isinstance(normalized, list)
+        else normalized
     )
 
 
-def canonical_node_diff(
+def semantic_diff(
     one: NodeSpec,
     two: NodeSpec,
     *,
     resolved_columns: list[ColumnSpec] | None,
     other_resolved_columns: list[ColumnSpec] | None,
 ) -> tuple[list[str], list[str]]:
-    """Compare two specs using their canonical semantic values."""
+    """Compare two specs using their normalized semantic values."""
     if one.node_type != two.node_type:
         return ["node_type"], []
 
@@ -236,12 +237,12 @@ def canonical_node_diff(
         if field == "display_name" and getattr(rendered_two, field) is None:
             continue
 
-        left = canonical_field_value(
+        left = normalize_field(
             rendered_one,
             field,
             resolved_columns=resolved_columns,
         )
-        right = canonical_field_value(
+        right = normalize_field(
             rendered_two,
             field,
             resolved_columns=other_resolved_columns,
@@ -252,13 +253,13 @@ def canonical_node_diff(
 
         if type(rendered_two).field_order_change_tier(field) == ChangeTier.NONE:
             continue
-        left_ordered = canonical_field_value(
+        left_ordered = normalize_field(
             rendered_one,
             field,
             resolved_columns=resolved_columns,
             preserve_order=True,
         )
-        right_ordered = canonical_field_value(
+        right_ordered = normalize_field(
             rendered_two,
             field,
             resolved_columns=other_resolved_columns,
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
index 557ae43de..adda049c4 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
@@ -15,10 +15,10 @@
     TransformSpec,
 )
 from datajunction_server.models.semantic_fingerprint import SemanticFingerprint
-from datajunction_server.semantic_fingerprints.canonical import (
-    canonical_field_value,
+from datajunction_server.semantic_fingerprints.normalization import (
     canonical_json,
-    canonical_value,
+    normalize_field,
+    normalize_value,
 )
 
 if TYPE_CHECKING:
@@ -70,11 +70,11 @@ def build_fingerprint(
     """Build a version 1 semantic fingerprint."""
     fingerprint_fields = semantic_fields(type(spec))
     for field in fingerprint_fields:
-        canonical_json(canonical_value(getattr(spec, field)))
+        canonical_json(normalize_value(getattr(spec, field)))
 
     rendered = spec.rendered_spec()
     fields = {
-        field: canonical_field_value(
+        field: normalize_field(
             rendered,
             field,
             resolved_columns=resolved_columns,
@@ -87,7 +87,7 @@ def build_fingerprint(
     }
     node_payload = {
         "domain": "datajunction/node-semantic",
-        "node_type": canonical_value(rendered.node_type),
+        "node_type": normalize_value(rendered.node_type),
         "fields": fields,
     }
     node_digest = hashlib.sha256(
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index 5b7644edf..e2c2d86a0 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -38,10 +38,10 @@
     eq_or_fallback,
     fold_change_tiers,
 )
-from datajunction_server.semantic_fingerprints.canonical import (
+from datajunction_server.semantic_fingerprints.normalization import (
     canonical_json,
-    canonical_sequence,
-    canonical_value,
+    normalize_sequence,
+    normalize_value,
 )
 from datajunction_server.semantic_fingerprints.v1 import semantic_fields
 from datajunction_server.models.materialization import (
@@ -1752,34 +1752,34 @@ def test_semantic_fingerprint_normalizes_empty_and_resolved_source_columns():
     ) == fingerprint(CubeSpec(name="cube", metrics=[], dimensions=[], filters=[]))
 
 
-def test_semantic_fingerprint_canonical_values_are_stable():
+def test_semantic_fingerprint_normalized_values_are_stable():
     first = {"outer": {"a": 1, "b": 2}, "value": 3}
     second = {"value": 3, "outer": {"b": 2, "a": 1}}
-    assert canonical_json(canonical_value(first)) == canonical_json(
-        canonical_value(second),
+    assert canonical_json(normalize_value(first)) == canonical_json(
+        normalize_value(second),
     )
     with pytest.raises(TypeError, match="string keys"):
-        canonical_value({1: "value"})
-    assert canonical_sequence(
+        normalize_value({1: "value"})
+    assert normalize_sequence(
         [1, 2],
         preserve_order=True,
-    ) != canonical_sequence(
+    ) != normalize_sequence(
         [2, 1],
         preserve_order=True,
     )
     with pytest.raises(TypeError, match="Unsupported"):
-        canonical_value({"bad": object()})
+        normalize_value({"bad": object()})
     with pytest.raises(ValueError, match="must be finite"):
-        canonical_value({"bad": float("nan")})
+        normalize_value({"bad": float("nan")})
 
 
 def test_semantic_fingerprint_normalizes_equivalent_numbers():
-    assert canonical_value({"value": 1}) == canonical_value({"value": 1.0})
-    assert canonical_value({"value": -0.0}) == canonical_value({"value": 0})
+    assert normalize_value({"value": 1}) == normalize_value({"value": 1.0})
+    assert normalize_value({"value": -0.0}) == normalize_value({"value": 0})
 
     sql_integer = TransformSpec(name="sql_number", query="SELECT 1")
     sql_integral_float = TransformSpec(name="sql_number", query="SELECT 1.0")
-    assert sql_integer.canonical_diff(sql_integral_float) == ([], [])
+    assert sql_integer.semantic_diff(sql_integral_float) == ([], [])
     assert fingerprint(sql_integer) == fingerprint(sql_integral_float)
 
 
@@ -1798,7 +1798,7 @@ def test_semantic_fingerprint_v1_projection_is_explicit(spec_type):
     assert set(semantic_fields(spec_type)) == current_major_fields
 
 
-def test_semantic_fingerprint_renders_prefixes_and_canonicalizes_sql():
+def test_semantic_fingerprint_renders_prefixes_and_normalizes_sql():
     from datajunction_server.models.dialect import Dialect
     from datajunction_server.sql.parsing.ast import render_for_dialect
 
@@ -1835,13 +1835,13 @@ def test_semantic_fingerprint_renders_prefixes_and_canonicalizes_sql():
     assert fingerprint(explicit) != fingerprint(implicit)
 
 
-def test_canonical_diff_and_fingerprint_share_change_rules():
+def test_semantic_diff_and_fingerprint_share_change_rules():
     original = TransformSpec(name="node", query="SELECT id AS value FROM source")
     formatted = TransformSpec(
         name="node",
         query=" SELECT  id AS value\nFROM source ",
     )
-    changed, reordered = original.canonical_diff(formatted)
+    changed, reordered = original.semantic_diff(formatted)
     assert (changed, reordered) == ([], [])
     assert TransformSpec.change_tier(changed, reordered) == ChangeTier.NONE
     assert fingerprint(original) == fingerprint(formatted)
@@ -1855,15 +1855,15 @@ def test_canonical_diff_and_fingerprint_share_change_rules():
     )
     source_changed = source.model_copy(deep=True)
     source_changed.columns[0].type = "string"
-    changed, reordered = source.canonical_diff(source_changed)
+    changed, reordered = source.semantic_diff(source_changed)
     assert (changed, reordered) == (["columns"], [])
     assert SourceSpec.change_tier(changed, reordered) == ChangeTier.MAJOR
     assert fingerprint(source) != fingerprint(source_changed)
-    assert source.canonical_diff(original) == (["node_type"], [])
+    assert source.semantic_diff(original) == (["node_type"], [])
 
     cube = CubeSpec(name="cube", metrics=["a", "b"], dimensions=[])
     reordered_cube = cube.model_copy(update={"metrics": ["b", "a", "a"]})
-    changed, reordered = cube.canonical_diff(reordered_cube)
+    changed, reordered = cube.semantic_diff(reordered_cube)
     assert (changed, reordered) == ([], ["metrics"])
     assert CubeSpec.change_tier(changed, reordered) == ChangeTier.MINOR
     assert fingerprint(cube) == fingerprint(reordered_cube)
@@ -1879,11 +1879,11 @@ def test_canonical_diff_and_fingerprint_share_change_rules():
         direction="neutral",
         unit={"kind": "currency", "code": "USD"},
     )
-    assert legacy_metric.canonical_diff(structured_metric) == ([], [])
+    assert legacy_metric.semantic_diff(structured_metric) == ([], [])
     changed, reordered = MetricSpec(
         name="metric",
         query="SELECT 1",
-    ).canonical_diff(legacy_metric)
+    ).semantic_diff(legacy_metric)
     assert (changed, reordered) == (["unit_enum"], [])
     assert MetricSpec.change_tier(changed, reordered) == ChangeTier.MINOR
 
@@ -2064,7 +2064,7 @@ def test_semantic_fingerprint_normalizes_primary_keys_and_cube_ordering():
         update={"required_dimensions": ["two", "one", "one"]},
     )
     assert metric == reordered_metric
-    assert metric.canonical_diff(reordered_metric) == ([], [])
+    assert metric.semantic_diff(reordered_metric) == ([], [])
     assert fingerprint(metric) == fingerprint(reordered_metric)
     assert MetricSpec.field_change_tier("columns") == ChangeTier.NONE
     assert fingerprint(cube) == fingerprint(

From 2fbcafcc8258f3cf775bcea8ea7b1c123d36440d Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 16:20:47 -0700
Subject: [PATCH 07/11] Cover semantic fingerprint edge cases

---
 .../tests/models/deployment_test.py           | 12 +++++++
 .../tests/sql/parsing/test_structural.py      | 31 ++++++++++++++++++-
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index e2c2d86a0..045f40ea2 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -3,6 +3,7 @@
 import subprocess
 import sys
 from datetime import date
+from decimal import Decimal
 
 import pytest
 from pydantic import ValidationError
@@ -1771,6 +1772,12 @@ def test_semantic_fingerprint_normalized_values_are_stable():
         normalize_value({"bad": object()})
     with pytest.raises(ValueError, match="must be finite"):
         normalize_value({"bad": float("nan")})
+    with pytest.raises(ValueError, match="must be finite"):
+        normalize_value(Decimal("NaN"))
+    assert normalize_value(True) is True
+    assert normalize_value(1.5) == 1.5
+    assert normalize_value(Decimal("1.0")) == 1
+    assert normalize_value(Decimal("1.50")) == {"decimal": "1.5"}
 
 
 def test_semantic_fingerprint_normalizes_equivalent_numbers():
@@ -1798,6 +1805,11 @@ def test_semantic_fingerprint_v1_projection_is_explicit(spec_type):
     assert set(semantic_fields(spec_type)) == current_major_fields
 
 
+def test_semantic_fingerprint_v1_rejects_unregistered_spec_type():
+    with pytest.raises(TypeError, match="No semantic fingerprint fields for NodeSpec"):
+        semantic_fields(NodeSpec)
+
+
 def test_semantic_fingerprint_renders_prefixes_and_normalizes_sql():
     from datajunction_server.models.dialect import Dialect
     from datajunction_server.sql.parsing.ast import render_for_dialect
diff --git a/datajunction-server/tests/sql/parsing/test_structural.py b/datajunction-server/tests/sql/parsing/test_structural.py
index 723d5ba5d..dc50964fb 100644
--- a/datajunction-server/tests/sql/parsing/test_structural.py
+++ b/datajunction-server/tests/sql/parsing/test_structural.py
@@ -1,10 +1,14 @@
 from dataclasses import dataclass
+from decimal import Decimal
 
 import pytest
 
 from datajunction_server.sql.parsing import ast
 from datajunction_server.sql.parsing.backends.antlr4 import parse
-from datajunction_server.sql.parsing.structural import serialize_ast
+from datajunction_server.sql.parsing.structural import (
+    _serialize_number_v1,
+    serialize_ast,
+)
 
 
 def test_serialize_ast_is_structural_and_numeric():
@@ -13,11 +17,34 @@ def test_serialize_ast_is_structural_and_numeric():
     typed = serialize_ast(
         parse("SELECT CAST(value AS DECIMAL(10, 2)) FROM source"),
     )
+    decimal = serialize_ast(ast.Number(Decimal("1.50")))
 
     assert integer == integral_float
     assert integer["type"] == "Query"
     assert "datajunction_server." not in str(integer)
     assert typed["type"] == "Query"
+    assert decimal["fields"]["value"] == {"decimal": "1.5"}
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (1, 1),
+        (1.5, 1.5),
+        (Decimal("1.0"), 1),
+        (Decimal("1.50"), {"decimal": "1.5"}),
+    ],
+)
+def test_serialize_number_v1(value, expected):
+    assert _serialize_number_v1(value) == expected
+
+
+def test_serialize_number_v1_rejects_booleans_and_nonfinite_values():
+    with pytest.raises(TypeError, match="Boolean values are not SQL numbers"):
+        _serialize_number_v1(True)
+    for value in (float("inf"), Decimal("NaN")):
+        with pytest.raises(ValueError, match="must be finite"):
+            _serialize_number_v1(value)
 
 
 def test_serialize_ast_rejects_unclassified_nodes_and_versions():
@@ -28,5 +55,7 @@ def __str__(self) -> str:
 
     with pytest.raises(TypeError, match="Unsupported structural SQL node: FutureNode"):
         serialize_ast(FutureNode())
+    with pytest.raises(TypeError, match="Unsupported structural SQL value: object"):
+        serialize_ast(ast.Name(name=object()))  # type: ignore[arg-type]
     with pytest.raises(ValueError, match="Unsupported structural SQL serialization"):
         serialize_ast(parse("SELECT 1"), version=2)

From 4401c87636bbc3a0db2e736b306d332d4144d662 Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 16:57:52 -0700
Subject: [PATCH 08/11] Represent unavailable semantic fingerprints explicitly

---
 .../datajunction_server/models/semantic_fingerprint.py      | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/datajunction-server/datajunction_server/models/semantic_fingerprint.py b/datajunction-server/datajunction_server/models/semantic_fingerprint.py
index aba526481..fc42e5131 100644
--- a/datajunction-server/datajunction_server/models/semantic_fingerprint.py
+++ b/datajunction-server/datajunction_server/models/semantic_fingerprint.py
@@ -1,5 +1,7 @@
 """Models and version constants for semantic fingerprints."""
 
+from typing import Literal, TypeAlias
+
 from pydantic import BaseModel, Field, field_validator
 
 
@@ -7,6 +9,7 @@
 SUPPORTED_SEMANTIC_FINGERPRINT_VERSIONS = frozenset(
     {LATEST_SEMANTIC_FINGERPRINT_VERSION},
 )
+UNKNOWN_SEMANTIC_FINGERPRINT: Literal["unknown"] = "unknown"
 
 
 class SemanticFingerprint(BaseModel):
@@ -25,3 +28,6 @@ def validate_version(cls, version: int) -> int:
         if version not in SUPPORTED_SEMANTIC_FINGERPRINT_VERSIONS:
             raise ValueError(f"Unsupported semantic fingerprint version: {version}")
         return version
+
+
+SemanticFingerprintValue: TypeAlias = SemanticFingerprint | Literal["unknown"]

From b6b3dbf6123fdd87d86d4eefc9db1251e1489890 Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 17:02:53 -0700
Subject: [PATCH 09/11] Compare unparseable queries without blocking deployment

---
 .../semantic_fingerprints/normalization.py    | 27 ++++++++++++-------
 .../tests/models/deployment_test.py           |  9 +++++++
 2 files changed, 26 insertions(+), 10 deletions(-)

diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
index 13e3d1462..f9628b92b 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
@@ -22,6 +22,7 @@
     SourceSpec,
 )
 from datajunction_server.models.node import MetricDirection
+from datajunction_server.sql.parsing.backends.exceptions import DJParseException
 
 
 def canonical_json(value: Any) -> str:
@@ -237,16 +238,22 @@ def semantic_diff(
         if field == "display_name" and getattr(rendered_two, field) is None:
             continue
 
-        left = normalize_field(
-            rendered_one,
-            field,
-            resolved_columns=resolved_columns,
-        )
-        right = normalize_field(
-            rendered_two,
-            field,
-            resolved_columns=other_resolved_columns,
-        )
+        try:
+            left = normalize_field(
+                rendered_one,
+                field,
+                resolved_columns=resolved_columns,
+            )
+            right = normalize_field(
+                rendered_two,
+                field,
+                resolved_columns=other_resolved_columns,
+            )
+        except DJParseException:
+            if field != "query":  # pragma: no cover
+                raise
+            left = rendered_one.rendered_query
+            right = rendered_two.rendered_query
         if left != right:
             changed_fields.append(field)
             continue
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index 045f40ea2..31f81d9d8 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -1900,6 +1900,15 @@ def test_semantic_diff_and_fingerprint_share_change_rules():
     assert MetricSpec.change_tier(changed, reordered) == ChangeTier.MINOR
 
 
+def test_semantic_diff_compares_unparseable_queries_as_raw_sql():
+    original = TransformSpec(name="node", query="SELECT (")
+    same = TransformSpec(name="node", query="SELECT (")
+    changed = TransformSpec(name="node", query="SELECT )")
+
+    assert original.semantic_diff(same) == ([], [])
+    assert original.semantic_diff(changed) == (["query"], [])
+
+
 @pytest.mark.parametrize(
     ("field", "value"),
     [

From dbd9420e5f1fa478216bc3ec731f575884526cbf Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Tue, 1 Sep 2026 18:48:04 -0700
Subject: [PATCH 10/11] Remove redundant semantic fingerprint work

---
 .../semantic_fingerprints/normalization.py                 | 2 +-
 .../datajunction_server/semantic_fingerprints/v1.py        | 7 -------
 .../datajunction_server/sql/parsing/structural.py          | 7 +++----
 3 files changed, 4 insertions(+), 12 deletions(-)

diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
index f9628b92b..4ddc34e1c 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py
@@ -53,7 +53,7 @@ def normalize_value(value: Any) -> Any:
     if isinstance(value, float):
         if not math.isfinite(value):
             raise ValueError("Semantic fingerprint values must be finite")
-        if value == 0 or value.is_integer():
+        if value.is_integer():
             return int(value)
         return value
     if isinstance(value, Decimal):
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
index adda049c4..08cc05e83 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py
@@ -7,7 +7,6 @@
 from typing import TYPE_CHECKING
 
 from datajunction_server.models.deployment import (
-    ChangeTier,
     CubeSpec,
     DimensionSpec,
     MetricSpec,
@@ -69,18 +68,12 @@ def build_fingerprint(
 ) -> SemanticFingerprint:
     """Build a version 1 semantic fingerprint."""
     fingerprint_fields = semantic_fields(type(spec))
-    for field in fingerprint_fields:
-        canonical_json(normalize_value(getattr(spec, field)))
-
     rendered = spec.rendered_spec()
     fields = {
         field: normalize_field(
             rendered,
             field,
             resolved_columns=resolved_columns,
-            preserve_order=(
-                type(rendered).field_order_change_tier(field) == ChangeTier.MAJOR
-            ),
             structural_version=1,
         )
         for field in fingerprint_fields
diff --git a/datajunction-server/datajunction_server/sql/parsing/structural.py b/datajunction-server/datajunction_server/sql/parsing/structural.py
index cdec4986d..fd00945cd 100644
--- a/datajunction-server/datajunction_server/sql/parsing/structural.py
+++ b/datajunction-server/datajunction_server/sql/parsing/structural.py
@@ -1,15 +1,14 @@
 """Versioned structural serialization for parsed SQL."""
 
+import math
 from collections.abc import Callable
 from decimal import Decimal
 from enum import Enum
-import math
 from typing import Any
 
 from datajunction_server.sql.parsing.ast import Node
 from datajunction_server.sql.parsing.types import ColumnType
 
-
 _AST_NODE_TAGS_V1 = frozenset(
     {
         "Alias",
@@ -64,13 +63,13 @@
 )
 
 
-def _serialize_number_v1(value: int | float | Decimal) -> int | float | dict[str, str]:
+def _serialize_number_v1(value: float | Decimal) -> int | float | dict[str, str]:
     if isinstance(value, bool):
         raise TypeError("Boolean values are not SQL numbers")
     if isinstance(value, float):
         if not math.isfinite(value):
             raise ValueError("Structural SQL numbers must be finite")
-        if value == 0 or value.is_integer():
+        if value.is_integer():
             return int(value)
         return value
     if isinstance(value, Decimal):

From 3411fcfaa9e6c62597f35b2b628e138f50881114 Mon Sep 17 00:00:00 2001
From: Philip Weiss <pweiss@netflix.com>
Date: Wed, 2 Sep 2026 12:41:49 -0700
Subject: [PATCH 11/11] Move semantic fingerprint composition out of node specs

Fingerprint values depend on graph context, so keep local and composition builders internal until the graph evaluator owns parent resolution.
---
 .../datajunction_server/models/deployment.py  | 21 --------
 .../semantic_fingerprints/__init__.py         |  7 +--
 .../semantic_fingerprints/engine.py           | 21 ++++++--
 .../tests/models/deployment_test.py           | 51 +++++++++++--------
 4 files changed, 51 insertions(+), 49 deletions(-)

diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py
index 031f62bc7..fbc047ef5 100644
--- a/datajunction-server/datajunction_server/models/deployment.py
+++ b/datajunction-server/datajunction_server/models/deployment.py
@@ -40,10 +40,6 @@
     NodeType,
 )
 from datajunction_server.models.partition import Granularity, PartitionType
-from datajunction_server.models.semantic_fingerprint import (
-    LATEST_SEMANTIC_FINGERPRINT_VERSION,
-    SemanticFingerprint,
-)
 from datajunction_server.models.unit import (
     Unit,
     legacy_unit_to_structured,
@@ -701,23 +697,6 @@ def rendered_spec(self) -> "NodeSpec":
         rendered_json = json.dumps(raw).replace("${prefix}", prefix)
         return self.__class__.model_validate_json(rendered_json)
 
-    def semantic_fingerprint(
-        self,
-        version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION,
-        *,
-        parent_fingerprints: Iterable[SemanticFingerprint] = (),
-        resolved_columns: list[ColumnSpec] | None = None,
-    ) -> SemanticFingerprint:
-        """Return a fingerprint of this node's semantic definition."""
-        from datajunction_server.semantic_fingerprints import fingerprint_node
-
-        return fingerprint_node(
-            self,
-            version,
-            parent_fingerprints=parent_fingerprints,
-            resolved_columns=resolved_columns,
-        )
-
     def semantic_diff(
         self,
         other: "NodeSpec",
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
index 368e16814..05ca21edf 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py
@@ -1,8 +1,5 @@
 """Semantic fingerprint construction and comparison."""
 
-from datajunction_server.semantic_fingerprints.engine import (
-    fingerprint_node,
-    semantic_diff,
-)
+from datajunction_server.semantic_fingerprints.engine import semantic_diff
 
-__all__ = ["fingerprint_node", "semantic_diff"]
+__all__ = ["semantic_diff"]
diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/engine.py b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
index f4cbc1469..5da8beeb2 100644
--- a/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
+++ b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py
@@ -23,14 +23,14 @@
 }
 
 
-def fingerprint_node(
+def compose_node_fingerprint(
     spec: NodeSpec,
     version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION,
     *,
-    parent_fingerprints: Iterable[SemanticFingerprint] = (),
+    parent_fingerprints: Iterable[SemanticFingerprint],
     resolved_columns: list[ColumnSpec] | None = None,
 ) -> SemanticFingerprint:
-    """Return a fingerprint of a node's semantic definition."""
+    """Compose a node fingerprint from its definition and parent fingerprints."""
     builder = _BUILDERS.get(version)
     if builder is None:
         raise ValueError(f"Unsupported semantic fingerprint version: {version}")
@@ -41,6 +41,21 @@ def fingerprint_node(
     )
 
 
+def local_node_fingerprint(
+    spec: NodeSpec,
+    version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION,
+    *,
+    resolved_columns: list[ColumnSpec] | None = None,
+) -> SemanticFingerprint:
+    """Fingerprint a node definition without graph parents."""
+    return compose_node_fingerprint(
+        spec,
+        version,
+        parent_fingerprints=(),
+        resolved_columns=resolved_columns,
+    )
+
+
 def semantic_diff(
     one: NodeSpec,
     two: NodeSpec,
diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py
index 31f81d9d8..91507b62d 100644
--- a/datajunction-server/tests/models/deployment_test.py
+++ b/datajunction-server/tests/models/deployment_test.py
@@ -30,7 +30,6 @@
     PartitionSpec,
     PartitionType,
     PreAggSpec,
-    SemanticFingerprint,
     SourceSpec,
     TagSpec,
     TransformSpec,
@@ -39,17 +38,22 @@
     eq_or_fallback,
     fold_change_tiers,
 )
+from datajunction_server.models.materialization import (
+    CoverageSpec,
+    MaterializationStrategy,
+)
+from datajunction_server.models.node import MetricUnit, NodeMode, NodeType
+from datajunction_server.models.semantic_fingerprint import SemanticFingerprint
+from datajunction_server.semantic_fingerprints.engine import (
+    compose_node_fingerprint,
+    local_node_fingerprint,
+)
 from datajunction_server.semantic_fingerprints.normalization import (
     canonical_json,
     normalize_sequence,
     normalize_value,
 )
 from datajunction_server.semantic_fingerprints.v1 import semantic_fields
-from datajunction_server.models.materialization import (
-    CoverageSpec,
-    MaterializationStrategy,
-)
-from datajunction_server.models.node import MetricUnit, NodeMode, NodeType
 
 
 def test_source_spec():
@@ -1695,7 +1699,7 @@ def semantic_specs() -> dict[str, NodeSpec]:
 
 
 def fingerprint(spec: NodeSpec) -> SemanticFingerprint:
-    return spec.semantic_fingerprint()
+    return local_node_fingerprint(spec)
 
 
 GOLDEN_FINGERPRINTS = {
@@ -1720,9 +1724,10 @@ def test_semantic_fingerprint_is_independent_of_python_hash_seed():
     script = """
 from datajunction_server.api.main import app
 from datajunction_server.models.deployment import ColumnSpec, SourceSpec
+from datajunction_server.semantic_fingerprints.engine import local_node_fingerprint
 spec = SourceSpec(name="s", catalog="c", schema_="s", table="t",
     columns=[ColumnSpec(name="id", attributes=["z", "primary_key", "a"])])
-print(spec.semantic_fingerprint().digest)
+print(local_node_fingerprint(spec).digest)
 """
 
     def digest_for(seed):
@@ -1744,7 +1749,8 @@ def test_semantic_fingerprint_normalizes_empty_and_resolved_source_columns():
     duplicated = SourceSpec(**common, columns=[*columns, columns[0].model_copy()])
 
     assert fingerprint(unspecified) == fingerprint(empty)
-    assert unspecified.semantic_fingerprint(
+    assert local_node_fingerprint(
+        unspecified,
         resolved_columns=columns,
     ) == fingerprint(resolved)
     assert fingerprint(resolved) == fingerprint(duplicated)
@@ -2121,7 +2127,7 @@ def test_semantic_fingerprint_rejects_unknown_version():
     with pytest.raises(ValidationError):
         SemanticFingerprint(version=2, digest="a" * 64)
     with pytest.raises(ValueError, match="Unsupported semantic fingerprint version: 2"):
-        semantic_specs()["source"].semantic_fingerprint(version=2)
+        local_node_fingerprint(semantic_specs()["source"], version=2)
 
 
 def test_semantic_fingerprint_combines_sorted_parent_hashes():
@@ -2131,23 +2137,28 @@ def test_semantic_fingerprint_combines_sorted_parent_hashes():
     first_hash = fingerprint(first)
     second_hash = fingerprint(second)
 
-    expected = node.semantic_fingerprint(
+    expected = compose_node_fingerprint(
+        node,
         parent_fingerprints=[first_hash, second_hash],
     )
-    assert expected == node.semantic_fingerprint(
+    assert expected == compose_node_fingerprint(
+        node,
         parent_fingerprints=[second_hash, first_hash, first_hash],
     )
-    assert expected != node.semantic_fingerprint(
+    assert expected != compose_node_fingerprint(
+        node,
         parent_fingerprints=[
             first_hash,
-            SourceSpec(
-                name="second",
-                catalog="c",
-                schema_="s",
-                table="changed",
-            ).semantic_fingerprint(),
+            fingerprint(
+                SourceSpec(
+                    name="second",
+                    catalog="c",
+                    schema_="s",
+                    table="changed",
+                ),
+            ),
         ],
     )
     mismatched = SemanticFingerprint.model_construct(version=2, digest="b" * 64)
     with pytest.raises(ValueError, match="Parent fingerprint version"):
-        node.semantic_fingerprint(parent_fingerprints=[mismatched])
+        compose_node_fingerprint(node, parent_fingerprints=[mismatched])