diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py index 2bf7081e5..fbc047ef5 100644 --- a/datajunction-server/datajunction_server/models/deployment.py +++ b/datajunction-server/datajunction_server/models/deployment.py @@ -19,7 +19,6 @@ DJInvalidDeploymentConfig, DJInvalidInputException, ) -from datajunction_server.models.base import labelize from datajunction_server.models.dimensionlink import ( JoinCardinality, JoinType, @@ -473,10 +472,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,30 +519,17 @@ 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 __eq__(self, other: object) -> bool: - if not isinstance(other, DimensionJoinLinkSpec): - return False # pragma: no cover + def _comparison_key(self) -> tuple[Any, ...]: 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 + *super()._comparison_key(), + self.rendered_dimension_node, + self.join_type, + self.join_cardinality, + self.rendered_join_on, + self.node_column, + self.default_value, ) @@ -570,24 +559,14 @@ 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 __eq__(self, other: object) -> bool: - if not isinstance(other, DimensionReferenceLinkSpec): - return False + def _comparison_key(self) -> tuple[Any, ...]: 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 + *super()._comparison_key(), + self.rendered_dimension_node, + self.dimension_attribute, + self.node_column, ) @@ -642,8 +621,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) # that can skip expensive SQL parsing and validation @@ -716,6 +697,25 @@ def rendered_spec(self) -> "NodeSpec": rendered_json = json.dumps(raw).replace("${prefix}", prefix) return self.__class__.model_validate_json(rendered_json) + 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 normalized values as fingerprints.""" + from datajunction_server.semantic_fingerprints import ( + semantic_diff as compare_semantics, + ) + + return compare_semantics( + 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. @@ -768,6 +768,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.""" @@ -777,13 +782,31 @@ 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.""" + from datajunction_server.semantic_fingerprints.normalization import ( + annotation_contains_list, + ) + + 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 @@ -842,6 +865,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): @@ -864,21 +892,25 @@ 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 ""), + from datajunction_server.semantic_fingerprints.normalization import ( + normalize_dimension_links, ) + 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 normalize_dimension_links( + self.dimension_links, + preserve_order=False, + ) + == normalize_dimension_links( + other.dimension_links, + preserve_order=False, ) - and dimension_links_equal and set(self.primary_key or []) == set(other.primary_key or []) ) @@ -983,7 +1015,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 @@ -996,6 +1028,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. @@ -1037,7 +1073,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. @@ -1049,7 +1085,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 @@ -1076,14 +1112,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: @@ -1095,12 +1131,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.normalization import ( + normalize_sequence, + ) + return ( super().__eq__(other) and self.query_ast.compare(other.query_ast) - and (self.required_dimensions or []) == (other.required_dimensions or []) + and normalize_sequence( + self.rendered_required_dimensions, + preserve_order=False, + ) + == 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 @@ -1177,6 +1224,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") @@ -1298,15 +1347,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. - 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 + from datajunction_server.semantic_fingerprints.normalization import ( + normalize_cube_columns, + ) + + return normalize_cube_columns( + self.rendered_columns, + ) == normalize_cube_columns(other.rendered_columns) NodeUnion = Annotated[ @@ -1366,7 +1413,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) @@ -1667,6 +1714,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.normalization import normalize_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 @@ -1675,39 +1724,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 = 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 = normalize_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/datajunction_server/models/semantic_fingerprint.py b/datajunction-server/datajunction_server/models/semantic_fingerprint.py new file mode 100644 index 000000000..fc42e5131 --- /dev/null +++ b/datajunction-server/datajunction_server/models/semantic_fingerprint.py @@ -0,0 +1,33 @@ +"""Models and version constants for semantic fingerprints.""" + +from typing import Literal, TypeAlias + +from pydantic import BaseModel, Field, field_validator + + +LATEST_SEMANTIC_FINGERPRINT_VERSION = 1 +SUPPORTED_SEMANTIC_FINGERPRINT_VERSIONS = frozenset( + {LATEST_SEMANTIC_FINGERPRINT_VERSION}, +) +UNKNOWN_SEMANTIC_FINGERPRINT: Literal["unknown"] = "unknown" + + +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 + + +SemanticFingerprintValue: TypeAlias = SemanticFingerprint | Literal["unknown"] 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..05ca21edf --- /dev/null +++ b/datajunction-server/datajunction_server/semantic_fingerprints/__init__.py @@ -0,0 +1,5 @@ +"""Semantic fingerprint construction and comparison.""" + +from datajunction_server.semantic_fingerprints.engine import 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 new file mode 100644 index 000000000..5da8beeb2 --- /dev/null +++ b/datajunction-server/datajunction_server/semantic_fingerprints/engine.py @@ -0,0 +1,72 @@ +"""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.normalization import ( + semantic_diff as compare_semantics, +) +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 compose_node_fingerprint( + spec: NodeSpec, + version: int = LATEST_SEMANTIC_FINGERPRINT_VERSION, + *, + parent_fingerprints: Iterable[SemanticFingerprint], + resolved_columns: list[ColumnSpec] | None = None, +) -> SemanticFingerprint: + """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}") + return builder( + spec, + parent_fingerprints, + resolved_columns=resolved_columns, + ) + + +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, + *, + resolved_columns: list[ColumnSpec] | None = None, + other_resolved_columns: list[ColumnSpec] | None = None, +) -> tuple[list[str], list[str]]: + """Compare two specs using the same normalized values as fingerprints.""" + return compare_semantics( + one, + two, + resolved_columns=resolved_columns, + other_resolved_columns=other_resolved_columns, + ) diff --git a/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py new file mode 100644 index 000000000..4ddc34e1c --- /dev/null +++ b/datajunction-server/datajunction_server/semantic_fingerprints/normalization.py @@ -0,0 +1,278 @@ +"""Semantic value normalization 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 +from datajunction_server.sql.parsing.backends.exceptions import DJParseException + + +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 normalize_value(value: Any) -> Any: + """Convert supported values to deterministic JSON-compatible values.""" + if isinstance(value, Enum): + return normalize_value(value.value) + if isinstance(value, BaseModel): + 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: normalize_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [normalize_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.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 normalize_sequence( + values: Iterable[Any], + *, + preserve_order: bool, +) -> list[Any]: + """Normalize a sequence while removing duplicate semantic values.""" + unique: dict[str, Any] = {} + for value in values: + normalized = normalize_value(value) + unique.setdefault(canonical_json(normalized), normalized) + 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 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 + 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 normalize_columns( + columns: list[ColumnSpec] | None, + *, + compare_types: bool, +) -> list[Any]: + """Normalize authored and resolved node columns.""" + column_map = {column.name: column for column in columns or []} + normalized_columns = [] + for name in sorted(column_map): + column = column_map[name] + normalized = normalize_column(column, name, column.type, compare_types) + if not compare_types: + default = normalize_column(None, name, column.type, compare_types) + if normalized == default: + continue + normalized_columns.append(normalize_value(normalized)) + return normalized_columns + + +def normalize_dimension_links( + links: list[DimensionJoinLinkSpec | DimensionReferenceLinkSpec] | None, + *, + preserve_order: bool, +) -> list[Any]: + """Normalize dimension links by their semantic comparison key.""" + return normalize_sequence( + (link._comparison_key() for link in links or []), + preserve_order=preserve_order, + ) + + +def normalize_cube_columns(columns: list[ColumnSpec] | None) -> dict[str, Any]: + """Normalize the authored partition configuration of cube columns.""" + return { + column.name: normalize_value(column.partition) + for column in columns or [] + if column.partition + } + + +def normalize_field( + spec: NodeSpec, + field: str, + *, + resolved_columns: list[ColumnSpec] | None = None, + preserve_order: bool = False, + structural_version: int | None = None, +) -> Any: + """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 + + 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 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 normalize_dimension_links( + spec.dimension_links, + preserve_order=preserve_order, + ) + if field == "unit_enum" and isinstance(spec, MetricSpec): + return normalize_value(spec._normalized_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 = [] + + normalized = normalize_value(value) + return ( + normalize_sequence(normalized, preserve_order=preserve_order) + if isinstance(normalized, list) + else normalized + ) + + +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 normalized 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 + + 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 + + if type(rendered_two).field_order_change_tier(field) == ChangeTier.NONE: + continue + left_ordered = normalize_field( + rendered_one, + field, + resolved_columns=resolved_columns, + preserve_order=True, + ) + right_ordered = normalize_field( + 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/v1.py b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py new file mode 100644 index 000000000..08cc05e83 --- /dev/null +++ b/datajunction-server/datajunction_server/semantic_fingerprints/v1.py @@ -0,0 +1,99 @@ +"""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 ( + CubeSpec, + DimensionSpec, + MetricSpec, + SourceSpec, + TransformSpec, +) +from datajunction_server.models.semantic_fingerprint import SemanticFingerprint +from datajunction_server.semantic_fingerprints.normalization import ( + canonical_json, + normalize_field, + normalize_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)) + rendered = spec.rendered_spec() + fields = { + field: normalize_field( + rendered, + field, + resolved_columns=resolved_columns, + structural_version=1, + ) + for field in fingerprint_fields + } + node_payload = { + "domain": "datajunction/node-semantic", + "node_type": normalize_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..fd00945cd --- /dev/null +++ b/datajunction-server/datajunction_server/sql/parsing/structural.py @@ -0,0 +1,136 @@ +"""Versioned structural serialization for parsed SQL.""" + +import math +from collections.abc import Callable +from decimal import Decimal +from enum import Enum +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: 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.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/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": [], diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py index c185b06d0..91507b62d 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 decimal import Decimal import pytest from pydantic import ValidationError @@ -39,6 +43,17 @@ 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 def test_source_spec(): @@ -885,6 +900,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(): @@ -906,7 +927,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(): @@ -995,7 +1016,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"] @@ -1629,3 +1650,515 @@ 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 local_node_fingerprint(spec) + + +GOLDEN_FINGERPRINTS = { + "source": "71dcbc388988c2bdd850670427710384687b58565ee38ca392dc220adfed868d", + "transform": "978e692880c7bcfb1bd78ece85895a1ec1558e85377b064a8dac3f3719cff2a5", + "dimension": "f7b3c87a61fdadf9997432fd9334befdf43f2488874ef555e3f7d4c4ba86e3e1", + "metric": "a3fde7af5dbd00d194805af33fc213bca52244c7cdedf1f7363ec52d2f6d4116", + "cube": "9b0a56d974d1e3769bc2db94e2cfbae7a6a4839f664eebd2a4387ef112ceea81", +} + + +@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) + assert result.version == 1 + + +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(local_node_fingerprint(spec).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 local_node_fingerprint( + unspecified, + 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_normalized_values_are_stable(): + first = {"outer": {"a": 1, "b": 2}, "value": 3} + second = {"value": 3, "outer": {"b": 2, "a": 1}} + assert canonical_json(normalize_value(first)) == canonical_json( + normalize_value(second), + ) + with pytest.raises(TypeError, match="string keys"): + normalize_value({1: "value"}) + assert normalize_sequence( + [1, 2], + preserve_order=True, + ) != normalize_sequence( + [2, 1], + preserve_order=True, + ) + with pytest.raises(TypeError, match="Unsupported"): + 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(): + 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.semantic_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_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 + + 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_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.semantic_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.semantic_diff(source_changed) + assert (changed, reordered) == (["columns"], []) + assert SourceSpec.change_tier(changed, reordered) == ChangeTier.MAJOR + assert fingerprint(source) != fingerprint(source_changed) + 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.semantic_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.semantic_diff(structured_metric) == ([], []) + changed, reordered = MetricSpec( + name="metric", + query="SELECT 1", + ).semantic_diff(legacy_metric) + assert (changed, reordered) == (["unit_enum"], []) + 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"), + [ + ("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"], + ) + reordered_metric = metric.model_copy( + update={"required_dimensions": ["two", "one", "one"]}, + ) + assert metric == 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( + 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(ValidationError): + SemanticFingerprint(version=2, digest="a" * 64) + with pytest.raises(ValueError, match="Unsupported semantic fingerprint version: 2"): + local_node_fingerprint(semantic_specs()["source"], 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 = compose_node_fingerprint( + node, + parent_fingerprints=[first_hash, second_hash], + ) + assert expected == compose_node_fingerprint( + node, + parent_fingerprints=[second_hash, first_hash, first_hash], + ) + assert expected != compose_node_fingerprint( + node, + parent_fingerprints=[ + first_hash, + 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"): + compose_node_fingerprint(node, parent_fingerprints=[mismatched]) 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..dc50964fb --- /dev/null +++ b/datajunction-server/tests/sql/parsing/test_structural.py @@ -0,0 +1,61 @@ +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_number_v1, + 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"), + ) + 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(): + @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(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)