From 84f05629622746c94be3c3085ea45e1cca655501 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Sat, 1 Aug 2026 22:23:03 -0700 Subject: [PATCH 1/5] Make semantic join paths deterministic --- sidemantic/core/semantic_graph.py | 135 +++++++++++++++++++++++----- sidemantic/loaders.py | 8 +- sidemantic/validation.py | 10 ++- tests/test_loaders.py | 33 ++++++- tests/test_semantic_graph_errors.py | 133 ++++++++++++++++++++++++++- tests/test_validation.py | 47 ++++++++++ 6 files changed, 338 insertions(+), 28 deletions(-) diff --git a/sidemantic/core/semantic_graph.py b/sidemantic/core/semantic_graph.py index bf9f0a2b2..449149d92 100644 --- a/sidemantic/core/semantic_graph.py +++ b/sidemantic/core/semantic_graph.py @@ -1,5 +1,6 @@ """Semantic graph for managing models and relationships.""" +import re from collections import deque from dataclasses import dataclass from typing import Any @@ -36,6 +37,29 @@ def _custom_join_condition(sql: str | None) -> str | None: return None +def _normalized_join_condition(sql: str | None) -> str | None: + """Canonicalize simple equality conjunctions for path de-duplication. + + Reciprocal relationship declarations generate the same custom join once as + ``from.x = to.x`` and once as ``to.x = from.x``. Equality and AND are + commutative, so normalize those shapes while leaving opaque predicates + distinct. This prevents a declared reciprocal edge from looking like a + second semantic path without hiding genuinely different join predicates. + """ + if sql is None: + return None + conjuncts = re.split(r"\s+AND\s+", sql.strip(), flags=re.IGNORECASE) + normalized: list[str] = [] + for conjunct in conjuncts: + comparison = re.fullmatch(r"\s*(.+?)\s*=\s*(.+?)\s*", conjunct) + if comparison and not any(operator in conjunct for operator in ("!=", "<=", ">=", "<>")): + sides = sorted(re.sub(r"\s+", "", side) for side in comparison.groups()) + normalized.append("=".join(sides)) + else: + normalized.append(re.sub(r"\s+", " ", conjunct.strip())) + return " AND ".join(sorted(normalized)) + + @dataclass class JoinPath: """Represents a join between two models.""" @@ -59,6 +83,10 @@ def to_entity(self) -> str: return self.to_columns[0] if self.to_columns else "" +class AmbiguousJoinPathError(ValueError): + """Raised when two models have more than one equally short semantic join path.""" + + class SemanticGraph: """Semantic graph managing models, metrics, and join relationships. @@ -79,10 +107,12 @@ def __init__(self): self._version = 0 self._adjacency_dirty = True self._adjacency: dict[str, list[tuple[str, list[str], list[str], str, str | None]]] = {} + self._relationship_path_cache: dict[tuple[str, str, frozenset[str] | None], tuple[JoinPath, ...] | str] = {} def _mark_dirty(self) -> None: self._version += 1 self._adjacency_dirty = True + self._relationship_path_cache.clear() def add_model(self, model: Model) -> None: """Add a model to the graph. @@ -287,6 +317,8 @@ def build_adjacency(self) -> None: if not hasattr(self, "_adjacency"): self._adjacency = {} self._adjacency.clear() + self._relationship_path_cache.clear() + self._adjacency_dirty = False def add_edge( from_model: str, @@ -403,18 +435,24 @@ def invert_relationship(relationship_type: str) -> str: _reverse_custom_join_condition(custom_condition), ) - def find_relationship_path(self, from_model: str, to_model: str) -> list[JoinPath]: - """Find join path between two models using BFS. + def find_relationship_path( + self, from_model: str, to_model: str, query_models: set[str] | frozenset[str] | None = None + ) -> list[JoinPath]: + """Find the unique shortest join path between two models. Args: from_model: Source model name to_model: Target model name + query_models: Optional models already referenced by the query. If + equally short routes exist, a unique route whose intermediate + hops stay in this set is preferred. Returns: List of JoinPath objects representing the join sequence Raises: ValueError: If no join path exists + AmbiguousJoinPathError: If multiple equally short paths exist """ if from_model == to_model: return [] @@ -428,23 +466,48 @@ def find_relationship_path(self, from_model: str, to_model: str) -> list[JoinPat if to_model not in self.models: raise KeyError(f"Model {to_model} not found") - # BFS to find shortest path - queue = deque([(from_model, [])]) - visited = {from_model} + context = frozenset(query_models) if query_models else None + cache_key = (from_model, to_model, context) + cached = self._relationship_path_cache.get(cache_key) + if isinstance(cached, str): + if cached.startswith("Ambiguous join paths"): + raise AmbiguousJoinPathError(cached) + raise ValueError(cached) + if cached is not None: + return list(cached) + + queue = deque([(from_model, tuple(), frozenset({from_model}))]) + shortest_length: int | None = None + candidates: dict[tuple[object, ...], tuple[JoinPath, ...]] = {} + + def edge_sort_key(edge): + next_model, from_keys, to_keys, relationship_type, custom_condition = edge + return next_model, tuple(from_keys), tuple(to_keys), relationship_type, custom_condition or "" + + def path_signature(path: tuple[JoinPath, ...]) -> tuple[object, ...]: + return tuple( + ( + hop.from_model, + hop.to_model, + tuple(hop.from_columns) if hop.custom_condition is None else (), + tuple(hop.to_columns) if hop.custom_condition is None else (), + hop.relationship, + _normalized_join_condition(hop.custom_condition), + ) + for hop in path + ) while queue: - current, path = queue.popleft() - - if current not in self._adjacency: + current, path, visited = queue.popleft() + if shortest_length is not None and len(path) >= shortest_length: continue - for next_model, from_keys, to_keys, relationship_type, custom_condition in self._adjacency[current]: + for next_model, from_keys, to_keys, relationship_type, custom_condition in sorted( + self._adjacency.get(current, []), key=edge_sort_key + ): if next_model in visited: continue - - visited.add(next_model) - - new_path = path + [ + new_path = path + ( JoinPath( from_model=current, to_model=next_model, @@ -452,15 +515,47 @@ def find_relationship_path(self, from_model: str, to_model: str) -> list[JoinPat to_columns=to_keys, relationship=relationship_type, custom_condition=custom_condition, - ) - ] + ), + ) if next_model == to_model: - return new_path - - queue.append((next_model, new_path)) - - raise ValueError(f"No join path found between {from_model} and {to_model}") + shortest_length = len(new_path) + candidates[path_signature(new_path)] = new_path + elif shortest_length is None or len(new_path) < shortest_length: + queue.append((next_model, new_path, visited | {next_model})) + + if not candidates: + message = f"No join path found between {from_model} and {to_model}" + self._relationship_path_cache[cache_key] = message + raise ValueError(message) + + ordered_candidates = [candidates[key] for key in sorted(candidates, key=repr)] + if len(ordered_candidates) > 1 and context is not None: + + def out_of_query_hops(path: tuple[JoinPath, ...]) -> int: + return sum(1 for hop in path[:-1] if hop.to_model not in context) + + best_score = min(out_of_query_hops(path) for path in ordered_candidates) + preferred = [path for path in ordered_candidates if out_of_query_hops(path) == best_score] + if len(preferred) == 1: + selected = preferred[0] + self._relationship_path_cache[cache_key] = selected + return list(selected) + ordered_candidates = preferred + + if len(ordered_candidates) > 1: + routes = [" -> ".join([from_model, *(hop.to_model for hop in path)]) for path in ordered_candidates] + message = ( + f"Ambiguous join paths between {from_model} and {to_model}: " + + "; ".join(routes) + + ". Define a unique relationship route before querying these models." + ) + self._relationship_path_cache[cache_key] = message + raise AmbiguousJoinPathError(message) + + selected = ordered_candidates[0] + self._relationship_path_cache[cache_key] = selected + return list(selected) def find_all_models_for_query(self, dimensions: list[str], measures: list[str]) -> set[str]: """Find all models needed for a query. diff --git a/sidemantic/loaders.py b/sidemantic/loaders.py index 51be996f8..040151b6b 100644 --- a/sidemantic/loaders.py +++ b/sidemantic/loaders.py @@ -1508,9 +1508,13 @@ def _unincluded(m) -> bool: # Find if any of these tables exist for target in potential_targets: if target in models and target != model_name and not _unincluded(models[target]): - # Check if this relationship already exists + # A relationship declared on either side is authoritative. In + # particular, the target may join on a non-primary key, while + # convention-based inference would fabricate a conflicting + # edge to the target's primary key. existing = [r for r in model.relationships if r.name == target] - if not existing: + declared_reverse = [r for r in models[target].relationships if r.name == model_name] + if not existing and not declared_reverse: # Add many_to_one relationship model.relationships.append( Relationship(name=target, type="many_to_one", foreign_key=dimension.name) diff --git a/sidemantic/validation.py b/sidemantic/validation.py index 6985a24fa..71c028432 100644 --- a/sidemantic/validation.py +++ b/sidemantic/validation.py @@ -889,12 +889,16 @@ def _add_untranslated_dax_model_error(model_ref: str, model) -> None: # Check that all model pairs can be joined # Only check models that exist in the graph (errors for missing models already reported above) - valid_model_names = [m for m in model_names if m in graph.models] - model_list = list(valid_model_names) + from sidemantic.core.semantic_graph import AmbiguousJoinPathError + + model_list = sorted(m for m in model_names if m in graph.models) + query_model_set = frozenset(model_list) for i, model_a in enumerate(model_list): for model_b in model_list[i + 1 :]: try: - graph.find_relationship_path(model_a, model_b) + graph.find_relationship_path(model_a, model_b, query_models=query_model_set) + except AmbiguousJoinPathError as exc: + errors.append(str(exc)) except (ValueError, KeyError): # Catch both ValueError (no path) and KeyError (model doesn't exist) errors.append( diff --git a/tests/test_loaders.py b/tests/test_loaders.py index 8ae205fa1..5870d1abf 100644 --- a/tests/test_loaders.py +++ b/tests/test_loaders.py @@ -6,8 +6,8 @@ import pytest -from sidemantic import SemanticLayer -from sidemantic.loaders import load_from_directory +from sidemantic import Dimension, Model, Relationship, SemanticLayer +from sidemantic.loaders import _infer_relationships, load_from_directory def test_load_from_directory_does_not_require_antlr4_without_antlr_formats(tmp_path, monkeypatch): @@ -35,6 +35,35 @@ def blocked_antlr4_import(name, *args, **kwargs): assert "orders" in layer.graph.models +def test_inference_respects_reverse_relationship_on_non_primary_key(): + invoices = Model( + name="invoices", + table="invoices", + primary_key="id", + dimensions=[Dimension(name="account_id", type="categorical")], + ) + accounts = Model( + name="accounts", + table="accounts", + primary_key="id", + dimensions=[Dimension(name="external_id", type="categorical")], + relationships=[ + Relationship( + name="invoices", + type="one_to_many", + foreign_key="account_id", + primary_key="external_id", + ) + ], + ) + + _infer_relationships({"invoices": invoices, "accounts": accounts}) + + assert invoices.relationships == [] + assert len(accounts.relationships) == 1 + assert accounts.relationships[0].primary_key == "external_id" + + def test_load_from_directory_strict_raises_on_detected_parse_error(tmp_path): """Strict loading fails instead of returning a partial graph.""" (tmp_path / "good.yml").write_text( diff --git a/tests/test_semantic_graph_errors.py b/tests/test_semantic_graph_errors.py index 10d41c0f0..9d6b7f352 100644 --- a/tests/test_semantic_graph_errors.py +++ b/tests/test_semantic_graph_errors.py @@ -4,7 +4,7 @@ from sidemantic.core.metric import Metric from sidemantic.core.model import Model -from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.core.semantic_graph import AmbiguousJoinPathError, SemanticGraph from sidemantic.core.table_calculation import TableCalculation @@ -119,6 +119,137 @@ def test_find_path_no_relationship(): graph.find_relationship_path("orders", "customers") +def test_find_path_rejects_registration_order_independent_diamond_ambiguity(): + """Two equally short routes must fail instead of selecting the first declared edge.""" + from sidemantic.core.relationship import Relationship + + def build_graph(reverse: bool) -> SemanticGraph: + first = Relationship(name="c", type="many_to_one", foreign_key="c_id") + second = Relationship(name="b", type="many_to_one", foreign_key="b_id") + relationships = [first, second] if reverse else [second, first] + graph = SemanticGraph() + graph.add_model(Model(name="a", table="a", primary_key="id", relationships=relationships)) + graph.add_model( + Model( + name="b", + table="b", + primary_key="id", + relationships=[Relationship(name="d", type="many_to_one", foreign_key="d_id")], + ) + ) + graph.add_model( + Model( + name="c", + table="c", + primary_key="id", + relationships=[Relationship(name="d", type="many_to_one", foreign_key="d_id")], + ) + ) + graph.add_model(Model(name="d", table="d", primary_key="id")) + return graph + + messages = [] + for reverse in (False, True): + graph = build_graph(reverse) + with pytest.raises(AmbiguousJoinPathError) as exc: + graph.find_relationship_path("a", "d") + messages.append(str(exc.value)) + + assert messages[0] == messages[1] + assert "a -> b -> d" in messages[0] + assert "a -> c -> d" in messages[0] + + +def test_find_path_prefers_a_unique_direct_route_over_longer_alternatives(): + from sidemantic.core.relationship import Relationship + + graph = SemanticGraph() + graph.add_model( + Model( + name="a", + table="a", + primary_key="id", + relationships=[ + Relationship(name="b", type="many_to_one", foreign_key="b_id"), + Relationship(name="d", type="many_to_one", foreign_key="d_id"), + ], + ) + ) + graph.add_model( + Model( + name="b", + table="b", + primary_key="id", + relationships=[Relationship(name="d", type="many_to_one", foreign_key="d_id")], + ) + ) + graph.add_model(Model(name="d", table="d", primary_key="id")) + + path = graph.find_relationship_path("a", "d") + + assert [(hop.from_model, hop.to_model) for hop in path] == [("a", "d")] + + +def test_relationship_path_cache_is_cleared_when_adjacency_is_rebuilt(): + from sidemantic.core.relationship import Relationship + + graph = SemanticGraph() + source = Model(name="a", table="a", primary_key="id") + graph.add_model(source) + graph.add_model(Model(name="b", table="b", primary_key="id")) + + with pytest.raises(ValueError, match="No join path found"): + graph.find_relationship_path("a", "b") + assert graph._relationship_path_cache + + source.relationships.append(Relationship(name="b", type="many_to_one", foreign_key="b_id")) + graph.build_adjacency() + + assert not graph._relationship_path_cache + assert [(hop.from_model, hop.to_model) for hop in graph.find_relationship_path("a", "b")] == [("a", "b")] + + +def test_reciprocal_custom_joins_are_one_semantic_path(): + from sidemantic.core.relationship import Relationship + + graph = SemanticGraph() + graph.add_model( + Model( + name="a", + table="a", + primary_key="id", + relationships=[ + Relationship( + name="b", + type="many_to_one", + foreign_key="b_id", + sql="{from}.b_id = {to}.id AND {from}.tenant_id = {to}.tenant_id", + ) + ], + ) + ) + graph.add_model( + Model( + name="b", + table="b", + primary_key="id", + relationships=[ + Relationship( + name="a", + type="one_to_many", + foreign_key="b_id", + sql="{from}.tenant_id = {to}.tenant_id AND {from}.id = {to}.b_id", + ) + ], + ) + ) + + path = graph.find_relationship_path("a", "b") + + assert len(path) == 1 + assert path[0].custom_condition is not None + + def test_auto_register_time_comparison_metric(): """Test that time_comparison metrics are auto-registered at graph level.""" graph = SemanticGraph() diff --git a/tests/test_validation.py b/tests/test_validation.py index ca5b5603d..34fc3d458 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -227,6 +227,53 @@ def test_query_validation_no_join_path(layer): assert "'products'" in str(exc_info.value) +def test_query_validation_reports_ambiguous_join_routes(layer): + layer.add_model( + Model( + name="a", + table="a", + primary_key="id", + metrics=[Metric(name="total", agg="count")], + relationships=[ + Relationship(name="b", type="many_to_one", foreign_key="b_id"), + Relationship(name="c", type="many_to_one", foreign_key="c_id"), + ], + ) + ) + layer.add_model( + Model( + name="b", + table="b", + primary_key="id", + relationships=[Relationship(name="d", type="many_to_one", foreign_key="d_id")], + ) + ) + layer.add_model( + Model( + name="c", + table="c", + primary_key="id", + relationships=[Relationship(name="d", type="many_to_one", foreign_key="d_id")], + ) + ) + layer.add_model( + Model( + name="d", + table="d", + primary_key="id", + dimensions=[Dimension(name="label", type="categorical")], + ) + ) + + with pytest.raises(QueryValidationError) as exc_info: + layer.compile(metrics=["a.total"], dimensions=["d.label"]) + + message = str(exc_info.value) + assert "Ambiguous join paths between a and d" in message + assert "a -> b -> d" in message + assert "a -> c -> d" in message + + def test_query_validation_invalid_granularity(layer): """Test that invalid time granularities are rejected.""" layer.add_model( From a9139c32fb74ea1288bb5ac606b7ad58b49aae60 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Sat, 1 Aug 2026 22:45:23 -0700 Subject: [PATCH 2/5] Optimize semantic loading and validation --- sidemantic/adapters/atscale_sml.py | 3 +- sidemantic/adapters/bsl.py | 3 +- sidemantic/adapters/cube.py | 3 +- sidemantic/adapters/hex.py | 3 +- sidemantic/adapters/metricflow.py | 3 +- sidemantic/adapters/omni.py | 9 +- sidemantic/adapters/osi.py | 3 +- sidemantic/adapters/rill.py | 3 +- sidemantic/adapters/sidemantic.py | 3 +- sidemantic/adapters/snowflake.py | 3 +- sidemantic/adapters/superset.py | 3 +- sidemantic/adapters/thoughtspot.py | 3 +- sidemantic/config.py | 6 +- sidemantic/core/preagg_management.py | 3 +- sidemantic/core/sql_definitions.py | 4 +- sidemantic/loaders.py | 75 +++++++++++-- sidemantic/project.py | 4 +- sidemantic/rust_bridge.py | 7 +- sidemantic/validation.py | 86 +++++++++----- sidemantic/validation_runner.py | 26 +++-- sidemantic/yaml_compat.py | 22 ++++ tests/test_loading_validation_performance.py | 111 +++++++++++++++++++ 22 files changed, 310 insertions(+), 76 deletions(-) create mode 100644 sidemantic/yaml_compat.py create mode 100644 tests/test_loading_validation_performance.py diff --git a/sidemantic/adapters/atscale_sml.py b/sidemantic/adapters/atscale_sml.py index 70164a9cc..1191e58ef 100644 --- a/sidemantic/adapters/atscale_sml.py +++ b/sidemantic/adapters/atscale_sml.py @@ -17,6 +17,7 @@ from sidemantic.core.pre_aggregation import PreAggregation from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load _TIME_UNIT_MAP = { "year": "year", @@ -350,7 +351,7 @@ def _load_objects(self, files: list[Path]) -> dict[str, dict[str, dict[str, Any] for file_path in files: with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: continue diff --git a/sidemantic/adapters/bsl.py b/sidemantic/adapters/bsl.py index 94f3a85c9..d5ae76b08 100644 --- a/sidemantic/adapters/bsl.py +++ b/sidemantic/adapters/bsl.py @@ -31,6 +31,7 @@ from sidemantic.core.model import Model from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class BSLAdapter(BaseAdapter): @@ -92,7 +93,7 @@ def _parse_file(self, file_path: Path, graph: SemanticGraph) -> None: graph: Semantic graph to add models to """ with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: return diff --git a/sidemantic/adapters/cube.py b/sidemantic/adapters/cube.py index 591fabc34..1e269e13c 100644 --- a/sidemantic/adapters/cube.py +++ b/sidemantic/adapters/cube.py @@ -17,6 +17,7 @@ from sidemantic.core.semantic_graph import SemanticGraph from sidemantic.fidelity import record_import_note from sidemantic.sql.fragment import replace_outside_sql_protected, rewrite_sql_column_spans +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class CubeImportWarning(UserWarning): @@ -405,7 +406,7 @@ def _parse_file( pending_extends: Dict to track extends relationships (child -> parent) """ with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: return diff --git a/sidemantic/adapters/hex.py b/sidemantic/adapters/hex.py index 7dce5e25f..a7aba5706 100644 --- a/sidemantic/adapters/hex.py +++ b/sidemantic/adapters/hex.py @@ -10,6 +10,7 @@ from sidemantic.core.model import Model from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load_all as _yaml_safe_load_all class HexAdapter(BaseAdapter): @@ -59,7 +60,7 @@ def _parse_file(self, file_path: Path, graph: SemanticGraph) -> None: graph: Semantic graph to add models to """ with open(file_path) as f: - documents = yaml.safe_load_all(f) + documents = _yaml_safe_load_all(f) for data in documents: if not data or not isinstance(data, dict): diff --git a/sidemantic/adapters/metricflow.py b/sidemantic/adapters/metricflow.py index 72c8ce4c9..45881f983 100644 --- a/sidemantic/adapters/metricflow.py +++ b/sidemantic/adapters/metricflow.py @@ -11,6 +11,7 @@ from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph from sidemantic.fidelity import record_import_note +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class MetricFlowAdapter(BaseAdapter): @@ -100,7 +101,7 @@ def _parse_file(self, file_path: Path, graph: SemanticGraph) -> None: graph: Semantic graph to add models/metrics to """ with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: return diff --git a/sidemantic/adapters/omni.py b/sidemantic/adapters/omni.py index 209fed06c..e7d70931b 100644 --- a/sidemantic/adapters/omni.py +++ b/sidemantic/adapters/omni.py @@ -11,6 +11,7 @@ from sidemantic.core.model import Model from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class OmniAdapter(BaseAdapter): @@ -148,7 +149,7 @@ def _is_model_or_relationships_file(cls, path: Path) -> bool: def _load_relationships_list(relationships_file: Path) -> list[dict[str, Any]]: """Load a bare top-level list of joins from a relationships file.""" with open(relationships_file) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if data is None: return [] @@ -169,7 +170,7 @@ def _parse_view(self, file_path: Path) -> Model | None: Model instance or None """ with open(file_path) as f: - view = yaml.safe_load(f) + view = _yaml_safe_load(f) if not view or not isinstance(view, dict): return None @@ -556,7 +557,7 @@ def _parse_relationships(self, model_file: Path, graph: SemanticGraph) -> None: graph: Semantic graph to add relationships to """ with open(model_file) as f: - model_def = yaml.safe_load(f) + model_def = _yaml_safe_load(f) if not model_def or not isinstance(model_def, dict): return @@ -664,7 +665,7 @@ def _parse_topic(self, topic_file: Path, graph: SemanticGraph) -> None: graph: Semantic graph to add topic + relationships to """ with open(topic_file) as f: - topic_def = yaml.safe_load(f) + topic_def = _yaml_safe_load(f) if not topic_def or not isinstance(topic_def, dict): return diff --git a/sidemantic/adapters/osi.py b/sidemantic/adapters/osi.py index 6d851df5b..552325550 100644 --- a/sidemantic/adapters/osi.py +++ b/sidemantic/adapters/osi.py @@ -31,6 +31,7 @@ from sidemantic.core.model import Model from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load # Directories that hold generated/compiled artifacts rather than source models. # dbt writes a copy of the OSI document to ``target/`` on ``dbt compile``; parsing @@ -186,7 +187,7 @@ def _parse_file(self, file_path: Path, graph: SemanticGraph) -> None: data = json.loads(text) if text.strip() else None else: # In-development OSI profile ships as YAML (the default). - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: return diff --git a/sidemantic/adapters/rill.py b/sidemantic/adapters/rill.py index c27780e76..2be2c0247 100644 --- a/sidemantic/adapters/rill.py +++ b/sidemantic/adapters/rill.py @@ -14,6 +14,7 @@ from sidemantic.core.metric import Metric from sidemantic.core.model import Model from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class RillAdapter: @@ -218,7 +219,7 @@ def _parse_file(self, file_path: Path) -> Model | None: Model if the file is a metrics_view, None otherwise """ with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data or data.get("type") != "metrics_view": return None diff --git a/sidemantic/adapters/sidemantic.py b/sidemantic/adapters/sidemantic.py index d63b99ff5..6578af225 100644 --- a/sidemantic/adapters/sidemantic.py +++ b/sidemantic/adapters/sidemantic.py @@ -21,6 +21,7 @@ parse_sql_graph_definitions, parse_sql_models, ) +from sidemantic.yaml_compat import safe_load as _yaml_safe_load NATIVE_FORMAT_VERSION = 1 ROOT_FIELDS = { @@ -405,7 +406,7 @@ def parse(self, source: str | Path) -> SemanticGraph: # Substitute environment variables content = substitute_env_vars(content) - data = yaml.safe_load(content) + data = _yaml_safe_load(content) if not data: return graph diff --git a/sidemantic/adapters/snowflake.py b/sidemantic/adapters/snowflake.py index 04292d484..b587effd2 100644 --- a/sidemantic/adapters/snowflake.py +++ b/sidemantic/adapters/snowflake.py @@ -14,6 +14,7 @@ from sidemantic.core.segment import Segment from sidemantic.core.semantic_graph import SemanticGraph from sidemantic.sql.fragment import parse_sql_fragment, rewrite_sql_column_spans +from sidemantic.yaml_compat import safe_load as _yaml_safe_load def _qualify_columns(sql_expr: str) -> str: @@ -213,7 +214,7 @@ def _parse_file( definitions, applied after every file's tables are loaded. """ with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not data: return diff --git a/sidemantic/adapters/superset.py b/sidemantic/adapters/superset.py index 73e2adb37..05c456b80 100644 --- a/sidemantic/adapters/superset.py +++ b/sidemantic/adapters/superset.py @@ -11,6 +11,7 @@ from sidemantic.core.metric import Metric from sidemantic.core.model import Model from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load class SupersetAdapter(BaseAdapter): @@ -70,7 +71,7 @@ def _parse_dataset(self, file_path: Path) -> Model | None: Model instance or None """ with open(file_path) as f: - dataset = yaml.safe_load(f) + dataset = _yaml_safe_load(f) if not dataset: return None diff --git a/sidemantic/adapters/thoughtspot.py b/sidemantic/adapters/thoughtspot.py index 083152368..623931c6e 100644 --- a/sidemantic/adapters/thoughtspot.py +++ b/sidemantic/adapters/thoughtspot.py @@ -14,6 +14,7 @@ from sidemantic.core.model import Model from sidemantic.core.relationship import Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load _BUCKET_MAP = { "HOURLY": "hour", @@ -681,7 +682,7 @@ def parse(self, source: str | Path) -> SemanticGraph: def _parse_file(self, file_path: Path) -> Model | None: with open(file_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) if not isinstance(data, dict): return None diff --git a/sidemantic/config.py b/sidemantic/config.py index c6e0c60a7..64adb87d2 100644 --- a/sidemantic/config.py +++ b/sidemantic/config.py @@ -5,6 +5,8 @@ from pydantic import BaseModel, ConfigDict, Field +from sidemantic.yaml_compat import safe_load as _yaml_safe_load + class DuckDBConnection(BaseModel): """DuckDB connection configuration.""" @@ -322,10 +324,8 @@ def load_config(config_path: Path) -> SidemanticConfig: suffix = config_path.suffix.lower() if suffix in {".yaml", ".yml"}: - import yaml - with open(config_path) as f: - data = yaml.safe_load(f) + data = _yaml_safe_load(f) elif suffix == ".json": with open(config_path) as f: data = json.load(f) diff --git a/sidemantic/core/preagg_management.py b/sidemantic/core/preagg_management.py index 5f462b3ed..840de761f 100644 --- a/sidemantic/core/preagg_management.py +++ b/sidemantic/core/preagg_management.py @@ -13,6 +13,7 @@ from sidemantic.core.pre_aggregation import PreAggregation from sidemantic.core.preagg_recommender import PreAggRecommendation, PreAggregationRecommender +from sidemantic.yaml_compat import safe_load as _yaml_safe_load RefreshMode = Literal["full", "incremental", "merge", "engine"] @@ -159,7 +160,7 @@ def _load_model_documents( locations: dict[str, list[ModelDefinitionLocation]] = {} yaml_files = sorted({*directory.rglob("*.yml"), *directory.rglob("*.yaml")}) for path in yaml_files: - loaded = yaml.safe_load(path.read_text()) + loaded = _yaml_safe_load(path.read_text()) if loaded is None: continue if not isinstance(loaded, dict): diff --git a/sidemantic/core/sql_definitions.py b/sidemantic/core/sql_definitions.py index fba3c60bb..5e41f2d0f 100644 --- a/sidemantic/core/sql_definitions.py +++ b/sidemantic/core/sql_definitions.py @@ -5,7 +5,6 @@ from pathlib import Path import sqlglot -import yaml from sqlglot import exp from sidemantic.core.dialect import ( @@ -35,6 +34,7 @@ from sidemantic.core.relationship import Relationship from sidemantic.core.segment import Segment from sidemantic.sql.aggregation_detection import sql_has_aggregate +from sidemantic.yaml_compat import safe_load as _yaml_safe_load def _split_top_level(text: str, delimiter: str = ",") -> list[str]: @@ -733,7 +733,7 @@ def parse_sql_file_with_frontmatter_extended( sql_body = parts[2].strip() if frontmatter_text: - frontmatter = yaml.safe_load(frontmatter_text) or {} + frontmatter = _yaml_safe_load(frontmatter_text) or {} _, _, _, metrics, segments, parameters, pre_aggregations = _parse_sql_statements(sql_body) diff --git a/sidemantic/loaders.py b/sidemantic/loaders.py index 040151b6b..16cbdb7dd 100644 --- a/sidemantic/loaders.py +++ b/sidemantic/loaders.py @@ -11,6 +11,8 @@ import yaml from sidemantic.fidelity import record_import_note +from sidemantic.yaml_compat import safe_load as _yaml_safe_load +from sidemantic.yaml_compat import safe_load_all as _yaml_safe_load_all if TYPE_CHECKING: from sidemantic.core.semantic_layer import SemanticLayer @@ -131,6 +133,39 @@ def _drop_non_registerable_models( return kept +_PRUNED_DIR_NAMES = frozenset( + { + ".git", + ".hg", + ".svn", + ".venv", + "venv", + "node_modules", + "__pycache__", + ".pytest_cache", + ".ruff_cache", + ".mypy_cache", + ".tox", + ".direnv", + ".claude", + ".idea", + ".vscode", + } +) + + +def _scan_project_files(directory: Path) -> list[Path]: + """Walk a project once while pruning dependency, VCS, and cache trees.""" + import os + + files: list[Path] = [] + for root, dirnames, filenames in os.walk(directory): + dirnames[:] = sorted(name for name in dirnames if name not in _PRUNED_DIR_NAMES) + root_path = Path(root) + files.extend(root_path / name for name in sorted(filenames)) + return files + + def load_from_directory( layer: "SemanticLayer", directory: str | Path, @@ -192,19 +227,24 @@ def load_from_directory( # relationship targeting a template survives being OVERWRITTEN by a later same-name real model. template_target_names: set[str] = set() + # Reuse one pruned traversal for every project-level format probe and the + # main per-file format detector below. + project_files = [] if only_file is not None else _scan_project_files(directory) + # Project-level formats (SML/TMDL/Graphene) are directory-based; in single-file # mode (only_file set) skip their whole-directory scans and parse just that file. - if only_file is None and _try_load_sml(layer, directory, all_models): + if only_file is None and _try_load_sml(layer, directory, all_models, project_files): return # TMDL projects are folder-based. Parse a project root once instead of # treating each .tmdl file as an independent model. tmdl_root = None if only_file is None: + tmdl_files = [file for file in project_files if file.suffix.lower() == ".tmdl"] definition_dir = directory / "definition" - if definition_dir.is_dir() and list(definition_dir.rglob("*.tmdl")): + if definition_dir.is_dir() and any(definition_dir in file.parents for file in tmdl_files): tmdl_root = definition_dir - elif list(directory.rglob("*.tmdl")): + elif tmdl_files: tmdl_root = directory if tmdl_root: @@ -237,7 +277,7 @@ def load_from_directory( # to a physical table, and be registered as queryable -- CLI validate/queries would silently # target a fabricated table. Parse the whole tree once instead (mirrors the TMDL handling). lookml_root = None - if only_file is None and any(directory.rglob("*.lkml")): + if only_file is None and any(file.suffix.lower() == ".lkml" for file in project_files): lookml_root = directory if lookml_root: @@ -280,10 +320,17 @@ def load_from_directory( lookml_root = None if only_file is None: - _load_graphene_project(directory, all_models, all_metrics, all_parameters, strict=strict) + _load_graphene_project( + directory, + all_models, + all_metrics, + all_parameters, + project_files, + strict=strict, + ) # Find and parse all files (just the requested one in single-file mode). - scan_files = [only_file] if only_file is not None else directory.rglob("*") + scan_files = [only_file] if only_file is not None else project_files for file_path in scan_files: if not file_path.is_file(): continue @@ -564,13 +611,14 @@ def _load_graphene_project( all_models: dict, all_metrics: dict, all_parameters: dict, + project_files: list[Path], *, strict: bool, ) -> None: """Parse Graphene `.gsql` files together so project-level links resolve.""" from sidemantic.adapters.graphene import GrapheneAdapter - if not any(directory.rglob("*.gsql")): + if not any(file.suffix.lower() == ".gsql" for file in project_files): return adapter = GrapheneAdapter() @@ -666,7 +714,7 @@ def _looks_like_python_semantic_definition(file_path: Path) -> bool: def _load_yaml_mapping(content: str) -> dict: """Parse YAML content and return a mapping, or an empty mapping for scalar/list YAML.""" - data = yaml.safe_load(content) + data = _yaml_safe_load(content) return data if isinstance(data, dict) else {} @@ -792,7 +840,7 @@ def _looks_like_hex_yaml(content: str) -> bool: uses ``safe_load_all`` and returns True when any document is a Hex resource. """ try: - documents = list(yaml.safe_load_all(content)) + documents = list(_yaml_safe_load_all(content)) except Exception: return False return any(_is_hex_resource_mapping(doc) for doc in documents) @@ -1315,7 +1363,12 @@ def _try_load_python_file( return True -def _try_load_sml(layer: "SemanticLayer", directory: Path, all_models: dict) -> bool: +def _try_load_sml( + layer: "SemanticLayer", + directory: Path, + all_models: dict, + project_files: list[Path], +) -> bool: """Detect and load an AtScale SML repository. Returns True if SML was found.""" for catalog_name in ("catalog.yml", "catalog.yaml", "atscale.yml", "atscale.yaml"): candidate = directory / catalog_name @@ -1325,7 +1378,7 @@ def _try_load_sml(layer: "SemanticLayer", directory: Path, all_models: dict) -> _load_sml_directory(layer, directory, all_models) return True - for sml_file in list(directory.rglob("*.yml")) + list(directory.rglob("*.yaml")): + for sml_file in (file for file in project_files if file.suffix.lower() in (".yml", ".yaml")): try: content = sml_file.read_text() except Exception: diff --git a/sidemantic/project.py b/sidemantic/project.py index c953515ad..d13cebfe1 100644 --- a/sidemantic/project.py +++ b/sidemantic/project.py @@ -61,10 +61,10 @@ def _load_config_values(config_path: Path) -> dict[str, Any]: with config_path.open() as config_file: values = json.load(config_file) else: - import yaml + from sidemantic.yaml_compat import safe_load as _yaml_safe_load with config_path.open() as config_file: - values = yaml.safe_load(config_file) + values = _yaml_safe_load(config_file) return values if isinstance(values, dict) else {} diff --git a/sidemantic/rust_bridge.py b/sidemantic/rust_bridge.py index 719570b45..022f18a48 100644 --- a/sidemantic/rust_bridge.py +++ b/sidemantic/rust_bridge.py @@ -9,6 +9,7 @@ import yaml from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.yaml_compat import safe_load as _yaml_safe_load # Lambda-only PreAggregation fields absent from the sidemantic-rs YAML schema # (which uses deny_unknown_fields). Exclude them when dumping a model so Rust @@ -580,7 +581,7 @@ def validate_models_payload_with_rust( def validate_model_payload_with_rust(model_obj) -> bool: """Validate model payload shape via sidemantic-rs.""" rust_module = get_rust_module() - payload = yaml.safe_load(models_to_rust_yaml([model_obj], include_extends=False)) or {} + payload = _yaml_safe_load(models_to_rust_yaml([model_obj], include_extends=False)) or {} model_payload = (payload.get("models") or [{}])[0] model_yaml = yaml.safe_dump(model_payload, sort_keys=False) return bool(rust_module.validate_model_payload(model_yaml)) @@ -643,7 +644,7 @@ def inherited_passthrough(model_name: str) -> dict: return dict(models) resolved_yaml = rust_module.resolve_model_inheritance(models_yaml) - resolved_payload = yaml.safe_load(resolved_yaml) or [] + resolved_payload = _yaml_safe_load(resolved_yaml) or [] resolved_models = {} for model_data in resolved_payload: normalized_data = dict(model_data) @@ -691,7 +692,7 @@ def resolve_metric_inheritance_with_rust(metrics: dict[str, object]) -> dict[str return dict(metrics) resolved_yaml = rust_module.resolve_metric_inheritance(metrics_yaml) - resolved_payload = yaml.safe_load(resolved_yaml) or [] + resolved_payload = _yaml_safe_load(resolved_yaml) or [] resolved_metrics = {} for metric_data in resolved_payload: normalized_data = dict(metric_data) diff --git a/sidemantic/validation.py b/sidemantic/validation.py index 71c028432..86e471d99 100644 --- a/sidemantic/validation.py +++ b/sidemantic/validation.py @@ -731,45 +731,73 @@ def validate_metric(measure: "Metric", graph: "SemanticGraph") -> list[str]: return errors -def _check_circular_dependencies( - measure: "Metric", graph: "SemanticGraph", visited: set[str], path: list[str] | None = None -) -> list[str] | None: - """Check for circular dependencies in derived measures. +_ACYCLIC_MEMO = None # weakref.WeakKeyDictionary, initialized lazily - Args: - measure: Metric to check - graph: Semantic graph - visited: Set of visited measure names - path: Current dependency path - Returns: - List of measure names in circular path, or None if no cycle - """ - if path is None: - path = [] +def _acyclic_safe_set(graph: "SemanticGraph") -> set[str]: + """Return metrics proven acyclic for the graph's current version.""" + global _ACYCLIC_MEMO + import weakref - if measure.name in visited: - # Found a cycle - cycle_start = path.index(measure.name) - return path[cycle_start:] + [measure.name] + if _ACYCLIC_MEMO is None: + _ACYCLIC_MEMO = weakref.WeakKeyDictionary() + version = getattr(graph, "_version", 0) + entry = _ACYCLIC_MEMO.get(graph) + if entry is None or entry[0] != version: + entry = (version, set()) + _ACYCLIC_MEMO[graph] = entry + return entry[1] + +def _check_circular_dependencies( + measure: "Metric", + graph: "SemanticGraph", + visited: set[str] | None = None, + path: list[str] | None = None, +) -> list[str] | None: + """Find a derived-metric cycle with iterative, graph-versioned DFS. + + ``visited`` and ``path`` remain accepted for compatibility with callers of + the former recursive helper. + """ if measure.type != "derived": return None - visited.add(measure.name) - path.append(measure.name) + safe = _acyclic_safe_set(graph) + if measure.name in safe: + return None - dependencies = measure.get_dependencies(graph) - for dep_name in dependencies: + def resolve(name: str): try: - dep_measure = graph.get_metric(dep_name) - if dep_measure: - cycle = _check_circular_dependencies(dep_measure, graph, visited.copy(), path.copy()) - if cycle: - return cycle + return graph.get_metric(name) except KeyError: - # Dependency doesn't exist yet, skip circular check - pass + return None + + on_stack = {measure.name} + order = [measure.name] + frames = [(measure, iter(measure.get_dependencies(graph)))] + + while frames: + current, dependencies = frames[-1] + advanced = False + for dependency_name in dependencies: + dependency = resolve(dependency_name) + if dependency is None or dependency.type != "derived" or dependency.name in safe: + continue + if dependency.name in on_stack: + cycle_start = order.index(dependency.name) + return order[cycle_start:] + [dependency.name] + on_stack.add(dependency.name) + order.append(dependency.name) + frames.append((dependency, iter(dependency.get_dependencies(graph)))) + advanced = True + break + if not advanced: + frames.pop() + on_stack.discard(current.name) + if order and order[-1] == current.name: + order.pop() + safe.add(current.name) return None diff --git a/sidemantic/validation_runner.py b/sidemantic/validation_runner.py index 4039787b3..b8e262deb 100644 --- a/sidemantic/validation_runner.py +++ b/sidemantic/validation_runner.py @@ -19,6 +19,21 @@ def passed(self) -> bool: return not self.errors +def _find_orphaned_models(models: dict[str, object]) -> list[str]: + """Return models with neither outgoing nor incoming relationships in O(V+E).""" + incoming_targets = { + relationship.name + for source_name, source in models.items() + for relationship in source.relationships + if relationship.name != source_name + } + return [ + model_name + for model_name, model in models.items() + if not model.relationships and model_name not in incoming_targets + ] + + def validate_directory(directory: str | Path) -> ValidationReport: """Load and validate semantic layer definitions from a directory.""" directory = Path(directory) @@ -68,16 +83,7 @@ def validate_directory(directory: str | Path) -> ValidationReport: report.errors.extend(validate_metric(metric, layer.graph)) if len(layer.graph.models) > 1: - orphaned = [] - for model_name, model in layer.graph.models.items(): - has_outgoing = bool(model.relationships) - has_incoming = any( - any(rel.name == model_name for rel in other.relationships) - for other_name, other in layer.graph.models.items() - if other_name != model_name - ) - if not has_outgoing and not has_incoming: - orphaned.append(model_name) + orphaned = _find_orphaned_models(layer.graph.models) if orphaned: report.warnings.append(f"Orphaned models (no relationships): {', '.join(orphaned)}") diff --git a/sidemantic/yaml_compat.py b/sidemantic/yaml_compat.py new file mode 100644 index 000000000..9714d119d --- /dev/null +++ b/sidemantic/yaml_compat.py @@ -0,0 +1,22 @@ +"""Fast safe YAML loading with a pure-Python fallback.""" + +from __future__ import annotations + +from typing import IO, Any + +import yaml + +try: # pragma: no cover - depends on how PyYAML was built + from yaml import CSafeLoader as _SafeLoader +except ImportError: # pragma: no cover - Pyodide does not ship libyaml + from yaml import SafeLoader as _SafeLoader + + +def safe_load(stream: str | bytes | IO) -> Any: + """Load one YAML document using libyaml when it is available.""" + return yaml.load(stream, Loader=_SafeLoader) + + +def safe_load_all(stream: str | bytes | IO) -> Any: + """Load all YAML documents using libyaml when it is available.""" + return yaml.load_all(stream, Loader=_SafeLoader) diff --git a/tests/test_loading_validation_performance.py b/tests/test_loading_validation_performance.py new file mode 100644 index 000000000..334485bbd --- /dev/null +++ b/tests/test_loading_validation_performance.py @@ -0,0 +1,111 @@ +"""Focused complexity and compatibility coverage for project loading/validation.""" + +import os +from types import SimpleNamespace + +import pytest +import yaml + +from sidemantic import Metric, SemanticLayer, load_from_directory +from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.validation import _check_circular_dependencies +from sidemantic.validation_runner import _find_orphaned_models +from sidemantic.yaml_compat import safe_load, safe_load_all + + +def test_directory_load_reuses_one_pruned_walk(tmp_path, monkeypatch): + (tmp_path / "models.yml").write_text( + """ +models: + - name: orders + table: orders + primary_key: id + dimensions: + - name: id + type: numeric + metrics: + - name: count + agg: count +""" + ) + ignored = tmp_path / ".venv" / "models" + ignored.mkdir(parents=True) + (ignored / "broken.yml").write_text("models: [") + + real_walk = os.walk + walked_roots = [] + + def counted_walk(root, *args, **kwargs): + walked_roots.append(root) + return real_walk(root, *args, **kwargs) + + monkeypatch.setattr(os, "walk", counted_walk) + + layer = SemanticLayer() + load_from_directory(layer, tmp_path) + + assert set(layer.graph.models) == {"orders"} + assert walked_roots == [tmp_path] + + +def test_yaml_compat_uses_safe_c_loader_when_available(): + from sidemantic import yaml_compat + + assert safe_load("answer: 42") == {"answer": 42} + assert list(safe_load_all("a: 1\n---\nb: 2\n")) == [{"a": 1}, {"b": 2}] + if hasattr(yaml, "CSafeLoader"): + assert yaml_compat._SafeLoader is yaml.CSafeLoader + with pytest.raises(yaml.constructor.ConstructorError): + safe_load("!!python/object/apply:builtins.eval ['1 + 1']") + + +def test_orphan_detection_walks_each_relationship_collection_once(): + class CountingRelationships(list): + iterations = 0 + + def __iter__(self): + type(self).iterations += 1 + return super().__iter__() + + model_count = 4_000 + models = {f"model_{index}": SimpleNamespace(relationships=CountingRelationships()) for index in range(model_count)} + models["model_0"].relationships.append(SimpleNamespace(name="model_1")) + + orphaned = _find_orphaned_models(models) + + assert orphaned == [f"model_{index}" for index in range(2, model_count)] + assert CountingRelationships.iterations == model_count + + +def test_derived_cycle_validation_is_iterative_and_memoized(monkeypatch): + graph = SemanticGraph() + metric_count = 1_200 + for index in range(metric_count): + graph.add_metric(Metric(name=f"metric_{index}", type="derived", sql=f"metric_{index + 1}")) + + real_get_dependencies = Metric.get_dependencies + dependency_scans = 0 + + def counted_dependencies(self, *args, **kwargs): + nonlocal dependency_scans + dependency_scans += 1 + return real_get_dependencies(self, *args, **kwargs) + + monkeypatch.setattr(Metric, "get_dependencies", counted_dependencies) + + for metric in graph.metrics.values(): + assert _check_circular_dependencies(metric, graph, set()) is None + + assert dependency_scans == metric_count + + +def test_derived_cycle_memo_is_invalidated_by_graph_version(): + graph = SemanticGraph() + first = Metric(name="first", type="derived", sql="second") + graph.add_metric(first) + + assert _check_circular_dependencies(first, graph, set()) is None + + graph.add_metric(Metric(name="second", type="derived", sql="first")) + + assert _check_circular_dependencies(first, graph, set()) == ["first", "second", "first"] From 3d832e66dafea440790ff8f6c631b42ce2cc9080 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Sat, 1 Aug 2026 22:31:54 -0700 Subject: [PATCH 3/5] Deduplicate fanout metrics by entity rows --- sidemantic/sql/generator.py | 457 ++++++++++++++++++- tests/metrics/test_symmetric_aggs.py | 269 ++++++++++- tests/optimizations/test_pre_aggregations.py | 70 +++ 3 files changed, 776 insertions(+), 20 deletions(-) diff --git a/sidemantic/sql/generator.py b/sidemantic/sql/generator.py index 79684fe50..0d0e2bd7b 100644 --- a/sidemantic/sql/generator.py +++ b/sidemantic/sql/generator.py @@ -332,6 +332,12 @@ def _agg_sql_name(agg: str) -> str: "approx_count_distinct": "APPROX_COUNT_DISTINCT", }.get(agg, agg.upper()) + def _safe_divide_sql(self, numerator: str, denominator: str) -> str: + """Build fractional division without PostgreSQL integer truncation.""" + if self.dialect == "postgres": + return f"CAST(({numerator}) AS DOUBLE PRECISION) / NULLIF(({denominator}), 0)" + return f"({numerator}) / NULLIF({denominator}, 0)" + @staticmethod def _model_from_clause(model) -> str: if model.sql: @@ -1754,6 +1760,23 @@ def _classify_filters_for_pushdown( return pushdown_filters, main_query_filters, window_dim_filters + def _filter_references_metric(self, filter_expr: str, model_names: set[str]) -> bool: + """Return whether a filter references a metric owned by one of the models.""" + try: + parsed = _parse_fragment(filter_expr, self.dialect) + except Exception: + # Fail closed: an unclassifiable filter stays at the outer aggregate grain. + return True + for column in parsed.find_all(exp.Column): + if not column.table: + continue + clean_name = column.table.replace("_cte", "") + if clean_name in model_names: + model = self.graph.get_model(clean_name) + if model and model.get_metric(column.name): + return True + return False + def _extract_metric_filter_columns(self, metrics: list[str]) -> dict[str, set[str]]: """Extract columns referenced in metric-level filters and SQL expressions. @@ -2733,13 +2756,21 @@ def _generate_with_preaggregation( segment_filters = self._resolve_segments(segments or []) all_filters = (filters or []) + segment_filters - # Partition filters by model so sub-queries only get relevant filters. - # Cross-model filters (referencing models outside the sub-query) would - # produce invalid SQL referencing CTEs that don't exist. + # Query-level row filters define one population, so every child query must + # see them. Otherwise sibling metrics in the final row can describe different + # populations. Metric filters remain at the outer aggregate grain. all_model_names = set(metrics_by_model.keys()) pushdown_by_model, shared_filters, window_dim_filters = self._classify_filters_for_pushdown( all_filters, all_model_names ) + child_filters = [filter_expr for model_filters in pushdown_by_model.values() for filter_expr in model_filters] + outer_filters = [] + for filter_expr in shared_filters: + if self._filter_references_metric(filter_expr, all_model_names): + outer_filters.append(filter_expr) + else: + child_filters.append(filter_expr) + shared_filters = outer_filters # Generate a pre-aggregated CTE for each metric model preagg_ctes = [] @@ -2749,11 +2780,11 @@ def _generate_with_preaggregation( cte_name = f"{model_name}_preagg" cte_names.append(cte_name) - # Pass pushdown filters plus any window-dim filters for this model. + # Pass every query-level row filter plus window-dim filters for this model. # Window-dim filters are pushed into the model's sub-query (not the # outer preagg join) so the recursive generate() handles them in its # own outer WHERE, preserving the requested dimension grain. - model_filters = pushdown_by_model.get(model_name, []) + window_dim_filters.get(model_name, []) + model_filters = child_filters + window_dim_filters.get(model_name, []) # Generate sub-query for this model's metrics at the dimension grain # We call generate() recursively but it won't trigger pre-aggregation @@ -3044,6 +3075,361 @@ def replace_field(column: exp.Column, _source: str) -> str | None: return query + def _rewrite_having_filter(self, filter_expr: str, metric_expressions: dict[str, str]) -> str: + """Replace semantic metric references with their aggregate expressions.""" + try: + parsed = _parse_fragment(filter_expr, self.dialect) + except Exception: + return filter_expr + + for column in list(parsed.find_all(exp.Column)): + qualified_ref = f"{column.table.replace('_cte', '')}.{column.name}" if column.table else None + replacement_sql = metric_expressions.get(qualified_ref) if qualified_ref else None + if replacement_sql is None and not column.table: + replacement_sql = metric_expressions.get(column.name) + if replacement_sql is None: + continue + try: + column.replace(_parse_fragment(replacement_sql, self.dialect)) + except Exception: + continue + return parsed.sql(dialect=self.dialect) + + def _fanout_safe_metric_plan( + self, + metrics: list[str], + symmetric_agg_needed: dict[str, bool], + ) -> tuple[str, list[tuple[str, object, str | None]], list[tuple[str, object]]] | None: + """Plan exact entity-row aggregation for metrics owned by one fanned-out model.""" + planned_outputs: list[tuple[str, object, str | None]] = [] + leaf_metrics: dict[str, tuple[str, object]] = {} + metric_models: set[str] = set() + supported_aggs = { + "sum", + "avg", + "count", + "count_distinct", + "approx_count_distinct", + "min", + "max", + "median", + "stddev", + "stddev_pop", + "variance", + "variance_pop", + } + + def resolve_reference(reference: str, model_context: str | None) -> tuple[str | None, object] | None: + if "." not in reference and model_context: + local_metric = self.graph.get_model(model_context).get_metric(reference) + if local_metric is not None: + return model_context, local_metric + try: + resolved_model_name, resolved_metric = self.graph.resolve_metric_reference(reference) + except KeyError: + return None + if resolved_metric is None: + return None + return resolved_model_name, resolved_metric + + def collect_leaves(metric, metric_context: str | None, stack: set[int]) -> bool: + identity = id(metric) + if identity in stack: + return False + stack = {*stack, identity} + + if getattr(metric, "sql_is_complete", False): + if metric_context is None or not metric.sql: + return False + try: + parsed = _parse_fragment(metric.sql.replace("{model}.", "").replace("{model}", ""), self.dialect) + except Exception: + return False + if any( + column.table and column.table.replace("_cte", "") != metric_context + for column in parsed.find_all(exp.Column) + ): + return False + columns = self._complete_sql_columns(metric) + if metric.filters and not columns: + return False + canonical_ref = f"{metric_context}.{metric.name}" + leaf_metrics.setdefault(canonical_ref, (canonical_ref, metric)) + metric_models.add(metric_context) + return True + if metric.agg: + if metric_context is None or metric.agg not in supported_aggs: + return False + canonical_ref = f"{metric_context}.{metric.name}" + leaf_metrics.setdefault(canonical_ref, (canonical_ref, metric)) + metric_models.add(metric_context) + return True + if metric.type == "ratio": + if not metric.numerator or not metric.denominator: + return False + for dependency in (metric.numerator, metric.denominator): + resolved = resolve_reference(dependency, metric_context) + if resolved is None or not collect_leaves(resolved[1], resolved[0] or metric_context, stack): + return False + return True + if metric.type == "derived" or (not metric.type and metric.sql): + if not metric.sql or "__bsl_all(" in metric.sql or sql_has_aggregate(metric.sql, self.dialect): + return False + for dependency in metric.get_dependencies(self.graph, metric_context): + resolved = resolve_reference(dependency, metric_context) + if resolved is None or not collect_leaves(resolved[1], resolved[0] or metric_context, stack): + return False + return True + return False + + for metric_ref in metrics: + resolved = resolve_reference(metric_ref, None) + if resolved is None: + return None + resolved_model_name, metric = resolved + if not collect_leaves(metric, resolved_model_name, set()): + return None + planned_outputs.append((metric_ref, metric, resolved_model_name)) + + if len(metric_models) != 1: + return None + model_name = next(iter(metric_models)) + if not symmetric_agg_needed.get(model_name, False): + return None + + model = self.graph.get_model(model_name) + self._require_primary_key(model_name, model.primary_key_columns, "to isolate measures across a fan-out join") + return model_name, planned_outputs, list(leaf_metrics.values()) + + def _build_fanout_safe_select( + self, + base_model_name: str, + other_models: list[str], + parsed_dims: list[tuple[str, str | None]], + metric_model_name: str, + planned_outputs: list[tuple[str, object, str | None]], + leaf_metrics: list[tuple[str, object]], + filters: list[str] | None, + models_with_filters: set[str], + order_by: list[str] | None, + limit: int | None, + offset: int | None, + aliases: dict[str, str], + with_totals: bool, + ) -> str: + """Aggregate one model from exact DISTINCT dimension plus typed-PK rows.""" + dedup_alias = "__sidemantic_dedup" + dim_internal_names = [f"__sidemantic_dim_{idx}" for idx in range(len(parsed_dims))] + model = self.graph.get_model(metric_model_name) + pk_internal_names = [f"__sidemantic_pk_{idx}" for idx in range(len(model.primary_key_columns))] + metric_internal_names = [f"__sidemantic_metric_{idx}" for idx in range(len(leaf_metrics))] + complete_column_internal_names: dict[int, dict[str, str]] = {} + for leaf_idx, (_metric_ref, measure) in enumerate(leaf_metrics): + if getattr(measure, "sql_is_complete", False): + complete_column_internal_names[id(measure)] = { + column_name: f"__sidemantic_complete_{leaf_idx}_{column_idx}" + for column_idx, (column_name, _quoted) in enumerate(self._complete_sql_columns(measure)) + } + + inner_select_exprs: list[str] = [] + for (dim_ref, gran), internal_name in zip(parsed_dims, dim_internal_names, strict=True): + dim_model_name, dim_name = dim_ref.split(".", 1) + cte_col_name = f"{dim_name}__{gran}" if gran else dim_name + inner_select_exprs.append( + f"{self._cte_ref(dim_model_name, cte_col_name)} AS {self._quote_alias(internal_name)}" + ) + for pk_col, internal_name in zip(model.primary_key_columns, pk_internal_names, strict=True): + inner_select_exprs.append( + f"{self._cte_ref(metric_model_name, pk_col)} AS {self._quote_alias(internal_name)}" + ) + for (_metric_ref, measure), internal_name in zip(leaf_metrics, metric_internal_names, strict=True): + if getattr(measure, "sql_is_complete", False): + for column_name, complete_internal_name in complete_column_internal_names[id(measure)].items(): + source_name = self._complete_sql_raw_alias(measure.name, column_name) + inner_select_exprs.append( + f"{self._cte_ref(metric_model_name, source_name)} AS {self._quote_alias(complete_internal_name)}" + ) + continue + inner_select_exprs.append( + f"{self._cte_ref(metric_model_name, f'{measure.name}_raw')} AS {self._quote_alias(internal_name)}" + ) + + inner_query = ( + select(*inner_select_exprs).distinct().from_(self._quote_identifier(self._cte_name(base_model_name))) + ) + inner_query = self._add_join_paths_to_query(inner_query, base_model_name, other_models, models_with_filters) + where_filters, having_filters = self._split_where_having_filters( + filters or [], [base_model_name] + other_models + ) + inner_query = self._add_where_filters_to_query(inner_query, where_filters, [base_model_name] + other_models) + + field_names: dict[str, list[str]] = {} + for dim_ref, gran in parsed_dims: + dim_model_name, dim_name = dim_ref.split(".", 1) + field_name = f"{dim_name}__{gran}" if gran else dim_name + field_names.setdefault(field_name, []).append(dim_model_name) + for metric_ref, metric, metric_context in planned_outputs: + owner = metric_context or (metric_ref.split(".", 1)[0] if "." in metric_ref else "") + field_names.setdefault(metric.name, []).append(owner) + has_collision = {name: len(owners) > 1 for name, owners in field_names.items()} + + output_aliases: dict[str, str] = {} + outer_select_exprs: list[str] = [] + dedup_table = self._quote_identifier(dedup_alias) + for idx, (dim_ref, gran) in enumerate(parsed_dims): + dim_model_name, dim_name = dim_ref.split(".", 1) + base_alias = f"{dim_name}__{gran}" if gran else dim_name + full_ref = f"{dim_ref}__{gran}" if gran else dim_ref + alias = aliases.get(full_ref) + if alias is None: + alias = f"{dim_model_name}_{base_alias}" if has_collision.get(base_alias, False) else base_alias + internal_ref = f"{dedup_table}.{self._quote_identifier(dim_internal_names[idx])}" + outer_select_exprs.append(f"{internal_ref} AS {self._quote_alias(alias)}") + output_aliases[full_ref] = alias + output_aliases[dim_ref] = alias + output_aliases[base_alias] = alias + + def aggregate_entity_rows(measure, raw_ref: str) -> str: + agg = measure.agg + if agg == "count": + aggregate = f"COUNT({raw_ref})" + elif agg == "count_distinct": + aggregate = f"COUNT({raw_ref})" if not measure.sql else f"COUNT(DISTINCT {raw_ref})" + elif agg == "approx_count_distinct": + aggregate = f"COUNT({raw_ref})" if not measure.sql else f"APPROX_COUNT_DISTINCT({raw_ref})" + else: + aggregate = f"{self._agg_sql_name(agg)}({raw_ref})" + return self._wrap_with_fill_nulls(aggregate, measure) + + def complete_sql_over_entity_rows(measure) -> str: + formula = (measure.sql or "").replace("{model}.", "").replace("{model}", "") + try: + parsed = _parse_fragment(formula, self.dialect) + except Exception as exc: + raise ValueError(f"Complete SQL metric {measure.name} could not be parsed safely") from exc + column_names = complete_column_internal_names.get(id(measure), {}) + for column in parsed.find_all(exp.Column): + if column.name not in column_names: + raise ValueError(f"Complete SQL metric {measure.name} references unsupported column {column.sql()}") + internal_name = column_names[column.name] + column.set( + "this", exp.to_identifier(internal_name, quoted=not self._is_simple_identifier(internal_name)) + ) + column.set("table", exp.to_identifier(dedup_alias, quoted=not self._is_simple_identifier(dedup_alias))) + entity_formula = parsed.sql(dialect=self.dialect) + if parsed_dims and not sql_has_aggregate(measure.sql or "", self.dialect): + entity_formula = f"ANY_VALUE({entity_formula})" + return entity_formula + + leaf_aggregate_by_id: dict[int, str] = {} + for idx, (_leaf_ref, measure) in enumerate(leaf_metrics): + if getattr(measure, "sql_is_complete", False): + continue + raw_ref = f"{dedup_table}.{self._quote_identifier(metric_internal_names[idx])}" + leaf_aggregate_by_id[id(measure)] = aggregate_entity_rows(measure, raw_ref) + + def resolve_calculation_reference(reference: str, model_context: str | None): + if "." not in reference and model_context: + local_metric = self.graph.get_model(model_context).get_metric(reference) + if local_metric is not None: + return model_context, local_metric + try: + return self.graph.resolve_metric_reference(reference) + except KeyError as exc: + raise ValueError(f"Metric {reference} not found") from exc + + def build_calculated_metric(metric, model_context: str | None, stack: set[int] | None = None) -> str: + stack = set() if stack is None else stack + if id(metric) in stack: + raise ValueError(f"Circular metric dependency involving {metric.name}") + stack = {*stack, id(metric)} + if getattr(metric, "sql_is_complete", False): + return complete_sql_over_entity_rows(metric) + if metric.agg: + if id(metric) not in leaf_aggregate_by_id: + raise ValueError(f"Metric {metric.name} was not planned at the entity grain") + return leaf_aggregate_by_id[id(metric)] + if metric.type == "ratio": + if not metric.numerator or not metric.denominator: + raise ValueError(f"Ratio metric {metric.name} requires numerator and denominator") + numerator_context, numerator = resolve_calculation_reference(metric.numerator, model_context) + denominator_context, denominator = resolve_calculation_reference(metric.denominator, model_context) + numerator_sql = build_calculated_metric(numerator, numerator_context or model_context, stack) + denominator_sql = build_calculated_metric(denominator, denominator_context or model_context, stack) + return self._safe_divide_sql(numerator_sql, denominator_sql) + if metric.type == "derived" or (not metric.type and metric.sql): + if not metric.sql: + raise ValueError(f"Derived metric {metric.name} missing sql") + formula = metric.sql + dependencies = sorted(metric.get_dependencies(self.graph, model_context), key=len, reverse=True) + import re + + for dependency in dependencies: + dependency_context, dependency_metric = resolve_calculation_reference(dependency, model_context) + dependency_sql = build_calculated_metric( + dependency_metric, dependency_context or model_context, stack + ) + if "." in dependency: + qualified_pattern = r"\b" + re.escape(dependency) + r"\b" + if re.search(qualified_pattern, formula): + formula = re.sub(qualified_pattern, f"({dependency_sql})", formula) + else: + bare_name = dependency.split(".", 1)[1] + formula = re.sub( + r"(? list of model names diff --git a/tests/metrics/test_symmetric_aggs.py b/tests/metrics/test_symmetric_aggs.py index 9d4c489c3..3b3fc09d5 100644 --- a/tests/metrics/test_symmetric_aggs.py +++ b/tests/metrics/test_symmetric_aggs.py @@ -1,9 +1,11 @@ """Tests for symmetric aggregates (fan-out join handling).""" import duckdb +import pytest from sidemantic.core.model import Dimension, Metric, Model, Relationship from sidemantic.core.semantic_graph import SemanticGraph +from sidemantic.core.semantic_layer import UnsupportedMetricError from sidemantic.core.symmetric_aggregate import build_symmetric_aggregate_sql from sidemantic.sql.generator import SQLGenerator from tests.utils import fetch_rows @@ -326,6 +328,266 @@ def test_symmetric_aggregates_with_data(): conn.close() +def test_fanout_isolates_typed_entity_rows_for_double_sum_avg_and_nulls(): + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE raw_orders (id BIGINT PRIMARY KEY, amount DOUBLE)") + conn.execute("INSERT INTO raw_orders VALUES (1, 100.25), (2, 50.75), (3, NULL)") + conn.execute(""" + CREATE TABLE raw_items AS + SELECT * FROM (VALUES + (1, 1, 'paid'), (2, 1, 'paid'), (3, 2, 'paid'), (4, 3, 'null-only') + ) AS t(id, order_id, category) + """) + + graph = SemanticGraph() + graph.add_model( + Model( + name="orders", + table="raw_orders", + primary_key="id", + metrics=[ + Metric(name="revenue", agg="sum", sql="amount"), + Metric(name="average_order_value", agg="avg", sql="amount"), + ], + relationships=[Relationship(name="items", type="one_to_many", sql="id", foreign_key="order_id")], + ) + ) + graph.add_model( + Model( + name="items", + table="raw_items", + primary_key="id", + dimensions=[Dimension(name="category", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + + sql = SQLGenerator(graph).generate( + metrics=["orders.revenue", "orders.average_order_value"], + dimensions=["items.category"], + order_by=["items.category"], + ) + assert "HASH(" not in sql + assert "SELECT DISTINCT" in sql + assert conn.execute(sql).fetchall() == [("null-only", None, None), ("paid", 151.0, 75.5)] + + +def test_fanout_evaluates_complete_sql_over_deduplicated_entity_rows(): + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE raw_orders AS SELECT * FROM (VALUES (1, 100.0), (2, 200.0)) t(id, amount)") + conn.execute( + "CREATE TABLE raw_items AS " + "SELECT * FROM (VALUES (1, 1, 'all'), (2, 1, 'all'), (3, 2, 'all')) t(id, order_id, category)" + ) + + graph = SemanticGraph() + graph.add_model( + Model( + name="orders", + table="raw_orders", + primary_key="id", + metrics=[ + Metric( + name="average_order_value", + sql="SUM({model}.amount) / COUNT(*)", + sql_is_complete=True, + ), + Metric(name="opaque_order_count", sql="COUNT(*)", sql_is_complete=True), + ], + relationships=[Relationship(name="items", type="one_to_many", sql="id", foreign_key="order_id")], + ) + ) + graph.add_model( + Model( + name="items", + table="raw_items", + primary_key="id", + dimensions=[Dimension(name="category", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + + sql = SQLGenerator(graph).generate( + metrics=["orders.average_order_value", "orders.opaque_order_count"], + dimensions=["items.category"], + ) + assert "SELECT DISTINCT" in sql + assert conn.execute(sql).fetchall() == [("all", 150.0, 2)] + + +def test_fanout_rejects_filtered_zero_column_complete_sql(): + graph = SemanticGraph() + graph.add_model( + Model( + name="orders", + table="raw_orders", + primary_key="id", + metrics=[ + Metric( + name="completed_count", + sql="COUNT(*)", + sql_is_complete=True, + filters=["{model}.status = 'completed'"], + ) + ], + relationships=[Relationship(name="items", type="one_to_many", sql="id", foreign_key="order_id")], + ) + ) + graph.add_model( + Model( + name="items", + table="raw_items", + primary_key="id", + dimensions=[Dimension(name="category", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + + with pytest.raises(UnsupportedMetricError, match="cannot be evaluated safely"): + SQLGenerator(graph).generate(metrics=["orders.completed_count"], dimensions=["items.category"]) + + +def test_fanout_typed_composite_keys_do_not_collide_on_delimiters(): + conn = duckdb.connect(":memory:") + conn.execute(""" + CREATE TABLE raw_orders AS + SELECT * FROM (VALUES + ('a|b', 'c', 100.0::DOUBLE), ('a', 'b|c', 200.0::DOUBLE) + ) AS t(part_a, part_b, amount) + """) + conn.execute(""" + CREATE TABLE raw_items AS + SELECT * FROM (VALUES + (1, 'a|b', 'c', 'all'), (2, 'a|b', 'c', 'all'), (3, 'a', 'b|c', 'all') + ) AS t(id, part_a, part_b, category) + """) + + join_sql = "{from}.part_a = {to}.part_a AND {from}.part_b = {to}.part_b" + graph = SemanticGraph() + graph.add_model( + Model( + name="orders", + table="raw_orders", + primary_key=["part_a", "part_b"], + metrics=[Metric(name="revenue", agg="sum", sql="amount")], + relationships=[Relationship(name="items", type="one_to_many", sql=join_sql)], + ) + ) + graph.add_model( + Model( + name="items", + table="raw_items", + primary_key="id", + dimensions=[Dimension(name="category", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", sql=join_sql)], + ) + ) + + sql = SQLGenerator(graph).generate(metrics=["orders.revenue"], dimensions=["items.category"]) + assert conn.execute(sql).fetchall() == [("all", 300.0)] + assert "CONCAT(" not in sql + + +def test_filter_only_sibling_fanout_is_deduplicated_for_non_base_metric(): + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE customers AS SELECT * FROM (VALUES (1, 'east')) AS t(id, region)") + conn.execute("CREATE TABLE orders AS SELECT * FROM (VALUES (1, 1, 100), (2, 1, 50)) t(id, customer_id, amount)") + conn.execute(""" + CREATE TABLE tickets AS + SELECT * FROM (VALUES (1, 1, 'open'), (2, 1, 'open'), (3, 1, 'closed')) t(id, customer_id, kind) + """) + + graph = SemanticGraph() + graph.add_model( + Model( + name="customers", + table="customers", + primary_key="id", + dimensions=[Dimension(name="region", type="categorical")], + relationships=[ + Relationship(name="orders", type="one_to_many", sql="id", foreign_key="customer_id"), + Relationship(name="tickets", type="one_to_many", sql="id", foreign_key="customer_id"), + ], + ) + ) + graph.add_model( + Model( + name="orders", + table="orders", + primary_key="id", + metrics=[Metric(name="revenue", agg="sum", sql="amount")], + relationships=[Relationship(name="customers", type="many_to_one", foreign_key="customer_id")], + ) + ) + graph.add_model( + Model( + name="tickets", + table="tickets", + primary_key="id", + dimensions=[Dimension(name="kind", type="categorical")], + relationships=[Relationship(name="customers", type="many_to_one", foreign_key="customer_id")], + ) + ) + + sql = SQLGenerator(graph).generate( + metrics=["orders.revenue"], + dimensions=["customers.region"], + filters=["tickets.kind = 'open'"], + ) + assert conn.execute(sql).fetchall() == [("east", 150)] + assert "SELECT DISTINCT" in sql + + +def test_derived_and_ratio_metrics_reuse_fanout_safe_leaf_aggregates(): + conn = duckdb.connect(":memory:") + conn.execute("CREATE TABLE raw_orders AS SELECT * FROM (VALUES (1, 100.0), (2, 200.0)) t(id, amount)") + conn.execute( + "CREATE TABLE raw_items AS " + "SELECT * FROM (VALUES (1, 1, 'all'), (2, 1, 'all'), (3, 2, 'all')) t(id, order_id, category)" + ) + + graph = SemanticGraph() + graph.add_model( + Model( + name="orders", + table="raw_orders", + primary_key="id", + metrics=[ + Metric(name="revenue", agg="sum", sql="amount"), + Metric(name="order_count", agg="count"), + Metric(name="double_revenue", type="derived", sql="revenue * 2"), + ], + relationships=[Relationship(name="items", type="one_to_many", sql="id", foreign_key="order_id")], + ) + ) + graph.add_model( + Model( + name="items", + table="raw_items", + primary_key="id", + dimensions=[Dimension(name="category", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + graph.add_metric( + Metric( + name="average_order_value", + type="ratio", + numerator="orders.revenue", + denominator="orders.order_count", + ) + ) + + sql = SQLGenerator(graph).generate( + metrics=["orders.revenue", "orders.order_count", "orders.double_revenue", "average_order_value"], + dimensions=["items.category"], + ) + assert conn.execute(sql).fetchall() == [("all", 300.0, 2, 600.0, 150.0)] + assert "HASH(" not in sql + assert "metric_0_metric_0" not in sql + assert sql.count("SELECT DISTINCT") == 1 + + def test_preagg_grain_preserved_with_filters(): """Test that preagg subqueries preserve the requested dimension grain when filters are applied. @@ -402,13 +664,12 @@ def test_preagg_grain_preserved_with_filters(): assert len(dates) == len(set(dates)), f"Duplicate dimension keys in preagg result: {dates}" # Verify correct values: - # The orders filter (status='shipped') only constrains orders_preagg. - # items_preagg aggregates ALL items by order_date (joined via orders). - # 2024-01-01: revenue=100 (shipped), total_qty=18 (items for orders 1+2) + # A query-level filter scopes every child to the same population. + # 2024-01-01: revenue=100 and total_qty=8 (items for shipped order 1). # 2024-01-02: revenue=150 (shipped), total_qty=7 (items for order 3) rows_sorted = sorted(rows, key=lambda r: r[0]) assert rows_sorted[0][1] == 100 # 2024-01-01 revenue (shipped only) - assert rows_sorted[0][2] == 18 # 2024-01-01 total_qty (all items for date) + assert rows_sorted[0][2] == 8 assert rows_sorted[1][1] == 150 # 2024-01-02 revenue assert rows_sorted[1][2] == 7 # 2024-01-02 total_qty diff --git a/tests/optimizations/test_pre_aggregations.py b/tests/optimizations/test_pre_aggregations.py index 22877cd2f..a7672d911 100644 --- a/tests/optimizations/test_pre_aggregations.py +++ b/tests/optimizations/test_pre_aggregations.py @@ -2770,5 +2770,75 @@ def test_lambda_union_does_not_double_count_boundary_bucket(): assert by_day["2024-01-02"] == 500 # boundary bucket re-aggregated once from source, not 800 +def _multi_fact_filter_layer(): + from sidemantic import Relationship, SemanticLayer + + layer = SemanticLayer() + con = layer.adapter.conn + con.execute("CREATE TABLE orders (id INTEGER, region VARCHAR, status VARCHAR, amount DOUBLE)") + con.execute("INSERT INTO orders VALUES (1,'US','completed',100),(2,'EU','completed',200)") + con.execute("CREATE TABLE line_items (id INTEGER, order_id INTEGER, qty INTEGER)") + con.execute("INSERT INTO line_items VALUES (1,1,5),(2,2,7)") + layer.add_model( + Model( + name="orders", + table="orders", + primary_key="id", + dimensions=[ + Dimension(name="status", type="categorical"), + Dimension(name="region", type="categorical"), + ], + metrics=[Metric(name="revenue", agg="sum", sql="amount")], + relationships=[Relationship(name="line_items", type="one_to_many", foreign_key="order_id")], + ) + ) + layer.add_model( + Model( + name="line_items", + table="line_items", + primary_key="id", + dimensions=[Dimension(name="qty_d", sql="qty", type="numeric")], + metrics=[Metric(name="total_qty", agg="sum", sql="qty")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + return layer + + +def test_fanout_preagg_applies_query_filters_to_every_child(): + layer = _multi_fact_filter_layer() + + rows = layer.query( + metrics=["orders.revenue", "line_items.total_qty"], + dimensions=["orders.status"], + filters=["orders.region = 'US'"], + ).fetchall() + assert rows == [("completed", 100.0, 5)] + + rows = layer.query( + metrics=["orders.revenue", "line_items.total_qty"], + dimensions=["orders.status"], + ).fetchall() + assert rows == [("completed", 300.0, 12)] + + +def test_fanout_preagg_metric_filter_stays_on_outer_query(): + layer = _multi_fact_filter_layer() + + rows = layer.query( + metrics=["orders.revenue", "line_items.total_qty"], + dimensions=["orders.status"], + filters=["orders.revenue > 50"], + ).fetchall() + assert rows == [("completed", 300.0, 12)] + + rows = layer.query( + metrics=["orders.revenue", "line_items.total_qty"], + dimensions=["orders.status"], + filters=["orders.revenue > 500"], + ).fetchall() + assert rows == [] + + if __name__ == "__main__": pytest.main([__file__, "-v"]) From 028867515978b3ffa7333ef782b3e6310cd7a239 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Sat, 1 Aug 2026 22:47:57 -0700 Subject: [PATCH 4/5] Correct advanced metric window semantics --- sidemantic/sql/generator.py | 525 +++++++++++++++++------ tests/metrics/test_advanced.py | 129 +++++- tests/metrics/test_cohort.py | 36 ++ tests/metrics/test_non_additive_guard.py | 65 +-- 4 files changed, 586 insertions(+), 169 deletions(-) diff --git a/sidemantic/sql/generator.py b/sidemantic/sql/generator.py index 0d0e2bd7b..51446bd20 100644 --- a/sidemantic/sql/generator.py +++ b/sidemantic/sql/generator.py @@ -15,13 +15,6 @@ from sidemantic.sql.fragment import _column_is_bound_in_select, parse_sql_fragment, rewrite_sql_column_spans from sidemantic.validation import QueryValidationError -# Dialects whose SQL supports the QUALIFY clause natively. Semi-additive -# (non_additive_dimension) handling emits a QUALIFY into each affected model's CTE -# to keep only the last (or first) snapshot per group; dialects without QUALIFY -# (e.g. postgres, mysql) are rejected rather than served an equivalent subquery in -# this first cut. See SQLGenerator._plan_semi_additive. -_QUALIFY_DIALECTS = frozenset({"duckdb", "snowflake", "bigquery", "databricks", "spark", "clickhouse"}) - @lru_cache(maxsize=4096) def _quote_identifier_cached(name: str, dialect: str, is_simple: bool) -> str: @@ -124,7 +117,7 @@ def __init__( allow_non_additive_unsafe: When True, skip the semi-additive rewrite for metrics that declare a non_additive_dimension and aggregate them naively over ALL snapshots (over-aggregated, wrong results). By default the generator - implements semi-additive handling (QUALIFY last/first snapshot per group); + implements semi-additive handling with portable nested window markers; this flag opts back into the old, naive behavior explicitly (default: False) enforce_visibility: Reject references to fields declared ``public: false`` base_model: Optional model that must anchor the generated join graph. Explore @@ -179,22 +172,21 @@ def _plan_semi_additive( base_model_name: str, model_names: list[str], parsed_dims: list[tuple[str, str | None]] | None = None, - ) -> dict[str, tuple[str, str]]: + ) -> dict[str, dict[str, tuple[str, str, tuple[str, ...] | None]]]: """Plan semi-additive (non_additive_dimension) handling for a grouped query. A measure with ``non_additive_dimension`` must be aggregated over only the rows at the last (max) — or first (min) — value of that time dimension per group, so an account-balance snapshot is summed across accounts but NOT - double-counted across days. We implement this by emitting a QUALIFY into the - owning model's CTE (see ``_build_model_cte``) that keeps only those rows. + double-counted across days. A nested SELECT marks the matching snapshot for + each affected leaf without filtering rows used by sibling metrics. - Returns a mapping ``model_name -> (non_additive_dimension_name, window)`` for - the models that need a semi-additive QUALIFY. ``window`` is "min" or "max". + Returns ``model -> metric -> (dimension, window, groupings)``. Each metric + receives its own marker in a nested SELECT, so sibling additive metrics keep + their full row set and dialects without QUALIFY are supported. Raises ``UnsupportedMetricError`` only for cases we do NOT implement: - - a dialect without QUALIFY (postgres, mysql, ...), since we don't emit the - equivalent subquery/self-join in this first cut; and - the combination of semi-additive handling with fan-out symmetric aggregation for the SAME model. These two rewrites don't compose safely: symmetric aggregation deduplicates fan-out rows via SUM(DISTINCT hash+value) at the @@ -209,7 +201,7 @@ def _plan_semi_additive( hatch: it now means "skip correct handling, aggregate naively", not "opt out of a guard". """ - plan: dict[str, tuple[str, str]] = {} + plan: dict[str, dict[str, tuple[str, str, tuple[str, ...] | None]]] = {} # Collect requested semi-additive measures and their owning models. Traverse metric # dependencies so a graph-level/derived metric that WRAPS a model measure declaring @@ -245,18 +237,8 @@ def _collect_semi_additive(reference: str) -> None: # Lazy import to avoid a circular import (semantic_layer imports this module). from sidemantic.core.semantic_layer import UnsupportedMetricError - if self.dialect not in _QUALIFY_DIALECTS: - reference = semi_additive[0][0] - raise UnsupportedMetricError( - f"Metric '{reference}' declares non_additive_dimension, which is handled " - f"with a QUALIFY clause, but dialect '{self.dialect}' has no QUALIFY support. " - "Semi-additive handling is currently implemented only for " - f"{sorted(_QUALIFY_DIALECTS)}. Pre-aggregate this metric upstream, or pass " - "allow_non_additive_unsafe=True to query it anyway (naive, over-aggregated)." - ) - - # Fan-out detection: a model whose measures are computed with symmetric - # aggregation cannot also carry a semi-additive QUALIFY (see docstring). + # Preserve the existing fan-out guard: this focused snapshot path does not + # replace symmetric aggregation or its entity-isolation semantics. fanout = self._has_fanout_joins(base_model_name, [m for m in model_names if m != base_model_name]) for reference, metric in semi_additive: @@ -289,8 +271,8 @@ def _collect_semi_additive(reference: str) -> None: # If the query groups by the non-additive dimension at its RAW grain, each # group is already a single snapshot value, so the measure is additive within - # the group and no QUALIFY is needed. But a COARSER grain (e.g. day__month) - # still spans many snapshots per bucket, so it must keep the QUALIFY (partitioned + # the group and no snapshot marker is needed. But a COARSER grain (e.g. day__month) + # still spans many snapshots per bucket, so it keeps the marker (partitioned # by the truncated bucket) -- otherwise a month-end-balance-style query silently # sums every daily snapshot in the month. Only the raw-grain case is additive. grouped_by_non_additive_raw = False @@ -307,19 +289,7 @@ def _collect_semi_additive(reference: str) -> None: window = getattr(metric, "non_additive_window", "max") or "max" groupings = getattr(metric, "non_additive_window_groupings", None) proposed = (metric.non_additive_dimension, window, tuple(groupings) if groupings else None) - existing = plan.get(owning_model) - if existing is not None and existing != proposed: - # A single model CTE carries one QUALIFY, so two semi-additive metrics on the - # same model that need different (dimension, window) snapshots cannot both be - # served correctly here (e.g. opening balance with min and closing with max). - # Refuse rather than silently applying only the last plan. - raise UnsupportedMetricError( - f"Model '{owning_model}' is queried with conflicting semi-additive metrics: " - f"one needs non_additive ({existing[0]!r}, window={existing[1]!r}) and another " - f"needs ({proposed[0]!r}, window={proposed[1]!r}). A single model CTE can carry " - "only one snapshot rule. Query them in separate requests, or pre-aggregate upstream." - ) - plan[owning_model] = proposed + plan.setdefault(owning_model, {})[metric.name] = proposed return plan @@ -1079,12 +1049,9 @@ def generate( segments, ) - # Semi-additive (non_additive_dimension) metrics are handled below by injecting - # a QUALIFY into the owning model's CTE (see _plan_semi_additive / _build_model_cte). - # The plan (and any UnsupportedMetricError for unimplemented dialects or the - # semi-additive + symmetric-aggregate combination) is computed once model_names - # is known; a non-empty plan also forces the live CTE path (pre-aggregations do - # not model per-group last-snapshot filtering). + # Semi-additive (non_additive_dimension) metrics use nested per-leaf window + # markers. Planning happens once model_names is known; a non-empty plan forces + # the live CTE path because rollups do not model snapshot selection. # Pre-aggregations are materialized with UTC time buckets, so a timezone-bucketed # query cannot be served from a rollup without returning wrong (UTC) buckets. Force a @@ -1255,12 +1222,12 @@ def metric_needs_window(m): # Find all models needed for the query model_names = self._find_required_models(metrics, dimensions, filters) - # Plan semi-additive handling. Raises for unimplemented cases (non-QUALIFY - # dialect, or semi-additive combined with fan-out symmetric aggregation on the - # same model). A non-empty plan forces the live CTE path (below) so the QUALIFY - # can be injected; pre-aggregation routing is skipped for these queries. + # Plan semi-additive handling. A non-empty plan forces the live CTE path; + # pre-aggregation routing is skipped because rollups do not encode snapshots. semi_additive_plan = ( - self._plan_semi_additive(metrics, model_names[0], model_names, parsed_dims) if model_names else {} + self._plan_semi_additive(metrics, model_names[0], model_names, parsed_dims) + if model_names and not ungrouped + else {} ) if semi_additive_plan: use_preaggregations = False @@ -1288,9 +1255,8 @@ def metric_needs_window(m): # Check if we need symmetric aggregation (pre-aggregation approach) # This is needed when metrics come from different models at different join levels. - # Skip this fan-out path for semi-additive queries: _plan_semi_additive already - # rejected any semi-additive measure whose owning model is fanned out, so a - # surviving plan must be served by the standard CTE path (which injects QUALIFY). + # Skip the multi-fact split for semi-additive queries; the snapshot path below + # owns the per-leaf row sets and rejects unsupported fan-out combinations. if not semi_additive_plan and self._needs_preaggregation_for_fanout(metrics, dimensions): return self._cache_generate_result( cache_key, @@ -1433,6 +1399,7 @@ def metric_needs_window(m): ungrouped=ungrouped, aliases=aliases, with_totals=with_totals, + semi_additive_plan=semi_additive_plan, ) # Combine CTEs and main query @@ -2006,7 +1973,7 @@ def _build_model_cte( all_models: set[str] | None = None, metric_filter_columns: set[str] | None = None, ungrouped: bool = False, - semi_additive: tuple[str, str] | None = None, + semi_additive: dict[str, tuple[str, str, tuple[str, ...] | None]] | None = None, ) -> str: """Build CTE SQL for a model with optional filter pushdown. @@ -2019,10 +1986,8 @@ def _build_model_cte( all_models: All models in query (for determining if joins needed) metric_filter_columns: Columns needed for metric-level filters ungrouped: Whether the query is returning raw ungrouped rows - semi_additive: Optional ``(non_additive_dimension_name, window)`` for a - semi-additive measure owned by this model. When set, a QUALIFY is added - so only rows at the last (window="max") or first ("min") value of that - time dimension per group survive. See ``_plan_semi_additive``. + semi_additive: Per-measure snapshot plans owned by this model. Their + time/grouping dimensions are projected for the nested snapshot stage. Returns: CTE SQL string @@ -2037,14 +2002,13 @@ def _build_model_cte( model_name, dimensions, filters, order_by, metric_filter_columns ) - # Semi-additive handling needs the non_additive_dimension column projected into - # this CTE (aliased to its dimension name) so the QUALIFY below can reference it, - # even when the query does not group by it. Declared window_groupings must also be - # projected so the QUALIFY can partition per those dimensions. + # Snapshot and declared grouping dimensions must be projected even when the + # query does not group by them so the nested window stage can reference them. if semi_additive is not None: - needed_dimensions.add(semi_additive[0]) - for grouping_col in semi_additive[2] or (): - needed_dimensions.add(grouping_col) + for non_additive_dim, _window, groupings in semi_additive.values(): + needed_dimensions.add(non_additive_dim) + for grouping_col in groupings or (): + needed_dimensions.add(grouping_col) # Build SELECT columns select_cols = [] @@ -2450,55 +2414,6 @@ def collect_measures_from_metric(metric_ref: str, visited: set[str] | None = Non where_clause = f"\n WHERE {' AND '.join(processed_filters)}" - # Build QUALIFY clause for semi-additive (non_additive_dimension) measures. - # Keep only the rows at the last (window="max") or first ("min") value of the - # non-additive time dimension per group. Referencing the CTE's own SELECT aliases - # here is valid under QUALIFY in DuckDB and the other _QUALIFY_DIALECTS. - qualify_clause = "" - if semi_additive is not None: - non_additive_dim, window, groupings = semi_additive - time_col = self._quote_identifier(non_additive_dim) - - partition_aliases: list[str] = [] - seen_partition: set[str] = set() - - def _add_partition(alias: str) -> None: - if alias not in seen_partition: - seen_partition.add(alias) - partition_aliases.append(self._quote_identifier(alias)) - - if groupings: - # Declared window_groupings (MetricFlow): the snapshot is taken per these - # dimensions regardless of the query's grouping (e.g. balance-per-user). - for grouping_col in groupings: - _add_partition(grouping_col) - # A coarser grain of the non-additive dim requested by the query still scopes - # the snapshot to that bucket (month-end balance per user). - for dim_ref, gran in dimensions: - if dim_ref == f"{model_name}.{non_additive_dim}" and gran is not None: - _add_partition(f"{non_additive_dim}__{gran}") - else: - # No declared groupings: partition by the query's other grouping dimensions - # for THIS model (e.g. account/region), excluding the raw non-additive axis. - for dim_ref, gran in dimensions: - if not dim_ref.startswith(model_name + "."): - continue - dim_name = dim_ref.split(".", 1)[1] - if dim_name == non_additive_dim and gran is None: - # The non-additive time dim at its RAW grain is the window axis, never a - # partition key. A coarser grain of it (day__month) IS a partition key. - continue - _add_partition(f"{dim_name}__{gran}" if gran else dim_name) - - agg = "MAX" if window == "max" else "MIN" - if partition_aliases: - partition_sql = ", ".join(partition_aliases) - window_expr = f"{agg}({time_col}) OVER (PARTITION BY {partition_sql})" - else: - # No other grouping dimensions: one global last/first snapshot. - window_expr = f"{agg}({time_col}) OVER ()" - qualify_clause = f"\n QUALIFY {time_col} = {window_expr}" - # Build CTE if not select_cols: # A complete-SQL measure whose SQL references no columns (e.g. a bare COUNT(*)) @@ -2509,7 +2424,7 @@ def _add_partition(alias: str) -> None: select_str = ",\n ".join(select_cols) cte_sql = ( f"{self._quote_identifier(self._cte_name(model_name))} AS " - f"(\n SELECT\n {select_str}\n FROM {from_clause}{where_clause}{qualify_clause}\n)" + f"(\n SELECT\n {select_str}\n FROM {from_clause}{where_clause}\n)" ) return cte_sql @@ -3430,6 +3345,245 @@ def build_calculated_metric(metric, model_context: str | None, stack: set[int] | outer_query = outer_query.offset(offset) return outer_query.sql(dialect=self.dialect, pretty=True) + def _build_semi_additive_select( + self, + base_model_name: str, + other_models: list[str], + parsed_dims: list[tuple[str, str | None]], + metrics: list[str], + filters: list[str] | None, + models_with_filters: set[str], + order_by: list[str] | None, + limit: int | None, + offset: int | None, + aliases: dict[str, str], + with_totals: bool, + semi_additive_plan: dict[str, dict[str, tuple[str, str, tuple[str, ...] | None]]], + ) -> str: + """Aggregate snapshot leaves independently in a portable nested query.""" + from sidemantic.core.semantic_layer import UnsupportedMetricError + + if len(semi_additive_plan) != 1: + raise UnsupportedMetricError("Semi-additive metrics from multiple metric models are unsupported") + metric_model_name = next(iter(semi_additive_plan)) + model_plan = semi_additive_plan[metric_model_name] + leaf_metrics: dict[str, object] = {} + planned_outputs: list[tuple[str, object, str | None]] = [] + + def resolve(reference: str, context: str | None) -> tuple[str | None, object]: + if "." not in reference and context: + local = self.graph.get_model(context).get_metric(reference) + if local is not None: + return context, local + try: + owner, metric = self.graph.resolve_metric_reference(reference) + except KeyError as exc: + raise ValueError(f"Metric {reference} not found") from exc + if metric is None: + raise ValueError(f"Metric {reference} not found") + return owner, metric + + def collect_leaves(metric, context: str | None, stack: set[int]) -> None: + if id(metric) in stack: + raise ValueError(f"Circular metric dependency involving {metric.name}") + stack = {*stack, id(metric)} + if metric.agg: + if context != metric_model_name: + raise UnsupportedMetricError("Semi-additive metrics cannot be combined with multiple metric models") + leaf_metrics.setdefault(metric.name, metric) + return + if metric.type == "ratio": + dependencies = [metric.numerator, metric.denominator] + elif metric.type == "derived" or (not metric.type and metric.sql): + dependencies = metric.get_dependencies(self.graph, context) + else: + dependencies = [] + if not dependencies or any(dependency is None for dependency in dependencies): + raise UnsupportedMetricError(f"Metric '{metric.name}' cannot be planned with semi-additive snapshots") + for dependency in dependencies: + dep_context, dep_metric = resolve(dependency, context) + collect_leaves(dep_metric, dep_context or context, stack) + + for metric_ref in metrics: + context, metric = resolve(metric_ref, None) + collect_leaves(metric, context, set()) + planned_outputs.append((metric_ref, metric, context)) + + dim_internal_names = [f"__sidemantic_dim_{idx}" for idx in range(len(parsed_dims))] + metric_internal_names = { + metric_name: f"__sidemantic_metric_{idx}" for idx, metric_name in enumerate(leaf_metrics) + } + snapshot_fields: dict[str, str] = {} + for non_additive_dim, _window, groupings in model_plan.values(): + for field_name in (non_additive_dim, *(groupings or ())): + snapshot_fields.setdefault(field_name, f"__sidemantic_snapshot_field_{len(snapshot_fields)}") + + inner_selects: list[str] = [] + for (dim_ref, gran), internal_name in zip(parsed_dims, dim_internal_names, strict=True): + dim_model, dim_name = dim_ref.split(".", 1) + source_name = f"{dim_name}__{gran}" if gran else dim_name + inner_selects.append(f"{self._cte_ref(dim_model, source_name)} AS {self._quote_alias(internal_name)}") + for metric_name, metric in leaf_metrics.items(): + internal_name = metric_internal_names[metric_name] + inner_selects.append( + f"{self._cte_ref(metric_model_name, f'{metric.name}_raw')} AS {self._quote_alias(internal_name)}" + ) + for field_name, internal_name in snapshot_fields.items(): + inner_selects.append( + f"{self._cte_ref(metric_model_name, field_name)} AS {self._quote_alias(internal_name)}" + ) + + inner_query = select(*inner_selects).from_(self._quote_identifier(self._cte_name(base_model_name))) + inner_query = self._add_join_paths_to_query(inner_query, base_model_name, other_models, models_with_filters) + where_filters, having_filters = self._split_where_having_filters( + filters or [], [base_model_name] + other_models + ) + inner_query = self._add_where_filters_to_query(inner_query, where_filters, [base_model_name] + other_models) + + rows_alias = "__sidemantic_rows" + rows_table = self._quote_identifier(rows_alias) + snapshot_selects = [ + f"{rows_table}.{self._quote_identifier(name)} AS {self._quote_alias(name)}" for name in dim_internal_names + ] + for metric_name, metric in leaf_metrics.items(): + internal_name = metric_internal_names[metric_name] + raw_ref = f"{rows_table}.{self._quote_identifier(internal_name)}" + snapshot = model_plan.get(metric.name) + if snapshot is not None: + non_additive_dim, window, groupings = snapshot + time_ref = f"{rows_table}.{self._quote_identifier(snapshot_fields[non_additive_dim])}" + partitions: list[str] = [] + + def add_partition(internal_name: str) -> None: + ref = f"{rows_table}.{self._quote_identifier(internal_name)}" + if ref not in partitions: + partitions.append(ref) + + if groupings: + for grouping in groupings: + add_partition(snapshot_fields[grouping]) + for idx, (dim_ref, gran) in enumerate(parsed_dims): + if dim_ref == f"{metric_model_name}.{non_additive_dim}" and gran is not None: + add_partition(dim_internal_names[idx]) + else: + for idx, (dim_ref, gran) in enumerate(parsed_dims): + if dim_ref == f"{metric_model_name}.{non_additive_dim}" and gran is None: + continue + add_partition(dim_internal_names[idx]) + + aggregate = "MAX" if window == "max" else "MIN" + partition_sql = f"PARTITION BY {', '.join(partitions)}" if partitions else "" + marker = f"{aggregate}({time_ref}) OVER ({partition_sql})" + raw_ref = f"CASE WHEN {time_ref} = {marker} THEN {raw_ref} ELSE NULL END" + snapshot_selects.append(f"{raw_ref} AS {self._quote_alias(internal_name)}") + + inner_sql = inner_query.sql(dialect=self.dialect, pretty=True) + snapshot_sql = ( + "SELECT\n " + + ",\n ".join(snapshot_selects) + + f"\nFROM (\n{inner_sql}\n) AS {self._quote_alias(rows_alias)}" + ) + snapshot_alias = "__sidemantic_snapshot" + snapshot_table = self._quote_identifier(snapshot_alias) + + leaf_aggregates: dict[int, str] = {} + for metric_name, metric in leaf_metrics.items(): + raw_ref = f"{snapshot_table}.{self._quote_identifier(metric_internal_names[metric_name])}" + if metric.agg == "count_distinct": + aggregate = f"COUNT(DISTINCT {raw_ref})" + elif metric.agg == "approx_count_distinct": + aggregate = f"APPROX_COUNT_DISTINCT({raw_ref})" + else: + aggregate = f"{self._agg_sql_name(metric.agg)}({raw_ref})" + leaf_aggregates[id(metric)] = self._wrap_with_fill_nulls(aggregate, metric) + + def build_metric(metric, context: str | None, stack: set[int] | None = None) -> str: + import re + + stack = set() if stack is None else stack + if id(metric) in stack: + raise ValueError(f"Circular metric dependency involving {metric.name}") + stack = {*stack, id(metric)} + if metric.agg: + return leaf_aggregates[id(metric)] + if metric.type == "ratio": + if not metric.numerator or not metric.denominator: + raise ValueError(f"Ratio metric {metric.name} requires numerator and denominator") + num_context, numerator = resolve(metric.numerator, context) + den_context, denominator = resolve(metric.denominator, context) + num_sql = build_metric(numerator, num_context or context, stack) + den_sql = build_metric(denominator, den_context or context, stack) + return f"({num_sql}) / NULLIF({den_sql}, 0)" + formula = metric.sql or "" + dependencies = sorted(metric.get_dependencies(self.graph, context), key=len, reverse=True) + for dependency in dependencies: + dep_context, dep_metric = resolve(dependency, context) + dep_sql = build_metric(dep_metric, dep_context or context, stack) + if "." in dependency and re.search(r"\b" + re.escape(dependency) + r"\b", formula): + formula = re.sub(r"\b" + re.escape(dependency) + r"\b", f"({dep_sql})", formula) + else: + bare_name = dependency.split(".", 1)[-1] + formula = re.sub(r"(? str: """Build main SELECT using SQLGlot builder API. @@ -3465,6 +3620,21 @@ def _build_main_select( SQL SELECT statement """ aliases = aliases or {} + if semi_additive_plan: + return self._build_semi_additive_select( + base_model_name, + other_models, + parsed_dims, + metrics, + filters, + models_with_filters, + order_by, + limit, + offset, + aliases, + with_totals, + semi_additive_plan, + ) # Detect if symmetric aggregates are needed symmetric_agg_needed = self._has_fanout_joins(base_model_name, other_models) @@ -3862,6 +4032,57 @@ def _calculate_lag_offset(self, comparison_type: str | None, time_granularity: s return 1 + @staticmethod + def _parse_period_interval(value: str | None) -> tuple[int, str] | None: + if not value: + return None + parts = value.strip().split() + if len(parts) != 2: + return None + try: + amount = int(parts[0]) + except ValueError: + return None + unit = parts[1].lower().rstrip("s") + if amount <= 0 or unit not in {"day", "week", "month", "quarter", "year"}: + return None + if unit == "quarter": + return amount * 3, "month" + return amount, unit + + def _comparison_period_interval(self, metric, time_granularity: str | None) -> tuple[int, str] | None: + explicit = self._parse_period_interval(metric.time_offset) + if explicit is not None: + return explicit + comparison_type = metric.comparison_type or "prior_period" + intervals = { + "dod": (1, "day"), + "wow": (1, "week"), + "mom": (1, "month"), + "qoq": (3, "month"), + "yoy": (1, "year"), + } + if comparison_type == "prior_period": + return (1, time_granularity) if time_granularity else None + return intervals.get(comparison_type) + + def _exact_period_lookup_sql( + self, + value_expr: str, + time_expr: str, + partition_clause: str, + interval: tuple[int, str] | None, + ) -> str | None: + """Lookup a value at an exact calendar offset using a value-based frame.""" + if interval is None or self.dialect not in {"duckdb", "postgres"}: + return None + amount, unit = interval + interval_sql = self._build_interval(str(amount), unit) + return ( + f"MAX({value_expr}) OVER ({partition_clause}ORDER BY {time_expr} " + f"RANGE BETWEEN {interval_sql} PRECEDING AND {interval_sql} PRECEDING)" + ) + def _offset_window_to_lag_rows(self, offset_window: str | None, time_granularity: str | None) -> int: """Convert a ratio offset_window (e.g. '3 months') to a row-based LAG offset. @@ -4662,7 +4883,7 @@ def _replace_model_placeholder(expr: str) -> str: else: dim_model_name = model_name dim_name = dim_ref_str - alias = dim_name + alias = f"{dim_name}__{granularity}" if granularity else dim_name if alias in entity_dim_aliases: continue # Already included @@ -5876,7 +6097,8 @@ def build_time_comparison_base_expression( if metric.window_expression: order_col = time_dim frame = metric.window_frame or "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW" - window_expr = f"{metric.window_expression} OVER (ORDER BY {order_col} {frame}) AS {metric_alias}" + window_value = f"{metric.window_expression} OVER (ORDER BY {order_col} {frame})" + window_expr = f"{self._wrap_with_fill_nulls(window_value, metric)} AS {metric_alias}" select_exprs.append(window_expr) cumulative_window_entries.append((window_expr, metric_alias)) continue @@ -5926,21 +6148,22 @@ def build_time_comparison_base_expression( grain = metric.grain_to_date partition = self._date_trunc(grain, time_dim) - window_expr = f"{agg_func}({base_col}) OVER (PARTITION BY {partition} ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}" + window_value = f"{agg_func}({base_col}) OVER (PARTITION BY {partition} ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)" elif metric.window: # Parse window (e.g., "7 days") window_parts = metric.window.split() if len(window_parts) == 2: num, unit = window_parts # For date-based windows, use RANGE - window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} RANGE BETWEEN INTERVAL '{num} {unit}' PRECEDING AND CURRENT ROW) AS {metric_alias}" + window_value = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} RANGE BETWEEN INTERVAL '{num} {unit}' PRECEDING AND CURRENT ROW)" else: # Fallback to rows - window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}" + window_value = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)" else: # Running total (unbounded window) - window_expr = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS {metric_alias}" + window_value = f"{agg_func}({base_col}) OVER (ORDER BY {time_dim} ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)" + window_expr = f"{self._wrap_with_fill_nulls(window_value, metric)} AS {metric_alias}" select_exprs.append(window_expr) cumulative_window_entries.append((window_expr, metric_alias)) @@ -6005,7 +6228,7 @@ def build_time_comparison_base_expression( time_dim = f"base.{dim_name}" if gran: time_dim = f"base.{dim_name}__{gran}" - time_dim_gran = gran + time_dim_gran = gran or dim.granularity break if not time_dim: @@ -6045,14 +6268,22 @@ def build_time_comparison_base_expression( else: lag_input_expr = metric_column(base_ref) - # Calculate LAG offset - lag_offset = self._calculate_lag_offset(metric.comparison_type, time_dim_gran) + exact_lookup = self._exact_period_lookup_sql( + lag_input_expr, + time_dim, + partition_clause, + ( + self._comparison_period_interval(metric, time_dim_gran) + if time_dim_gran or metric.time_offset + else None + ), + ) + if exact_lookup is None: + lag_offset = self._calculate_lag_offset(metric.comparison_type, time_dim_gran) + exact_lookup = f"LAG({lag_input_expr}, {lag_offset}) OVER ({partition_clause}ORDER BY {time_dim})" - # Add LAG for base metric (quote alias to handle dotted names) prev_value_alias = self._quote_alias(f"{m}_prev_value") - lag_selects.append( - f"LAG({lag_input_expr}, {lag_offset}) OVER ({partition_clause}ORDER BY {time_dim}) AS {prev_value_alias}" - ) + lag_selects.append(f"{exact_lookup} AS {prev_value_alias}") # Add LAG expressions for each offset ratio metric for m in offset_ratio_metrics: @@ -6103,13 +6334,21 @@ def build_time_comparison_base_expression( # Get denominator alias denom_alias = metric_ref_alias(metric.denominator) - # Add LAG for denominator - reference base.denom_alias since it's from inner query - # Quote alias to handle dotted names - lag_rows = self._offset_window_to_lag_rows(metric.offset_window, time_dim_gran) - prev_denom_alias = self._quote_alias(f"{m}_prev_denom") - lag_selects.append( - f"LAG(base.{sql_identifier(denom_alias)}, {lag_rows}) OVER ({partition_clause}ORDER BY {time_dim}) AS {prev_denom_alias}" + denominator_expr = f"base.{sql_identifier(denom_alias)}" + interval = self._parse_period_interval(metric.offset_window) if time_dim_gran else None + prior_denominator = self._exact_period_lookup_sql( + denominator_expr, + time_dim, + partition_clause, + interval, ) + if prior_denominator is None: + lag_rows = self._offset_window_to_lag_rows(metric.offset_window, time_dim_gran) + prior_denominator = ( + f"LAG({denominator_expr}, {lag_rows}) OVER ({partition_clause}ORDER BY {time_dim})" + ) + prev_denom_alias = self._quote_alias(f"{m}_prev_denom") + lag_selects.append(f"{prior_denominator} AS {prev_denom_alias}") # Build intermediate CTE - inner_query already has all the columns we need # We need to add "base." prefix since we're wrapping inner_query in a FROM (inner_query) AS base @@ -6141,14 +6380,15 @@ def build_time_comparison_base_expression( # Build calculation based on calculation type calc_type = metric.calculation or "percent_change" if calc_type == "difference": - expr = f"({base_alias} - {prev_value_col}) AS {final_alias}" + value_expr = f"({base_alias} - {prev_value_col})" elif calc_type == "percent_change": - expr = f"(({base_alias} - {prev_value_col}) / NULLIF({prev_value_col}, 0) * 100) AS {final_alias}" + value_expr = f"(({base_alias} - {prev_value_col}) / NULLIF({prev_value_col}, 0) * 100)" elif calc_type == "ratio": - expr = f"({base_alias} / NULLIF({prev_value_col}, 0)) AS {final_alias}" + value_expr = f"({base_alias} / NULLIF({prev_value_col}, 0))" else: raise ValueError(f"Unknown calculation type: {calc_type}") + expr = f"{self._wrap_with_fill_nulls(value_expr, metric)} AS {final_alias}" final_selects.append(expr) # Add offset ratio metrics @@ -6165,7 +6405,8 @@ def build_time_comparison_base_expression( final_alias = self._quote_alias(m) # Calculate ratio using the lagged value - offset_expr = f"{num_alias} / NULLIF({prev_denom_col}, 0) AS {final_alias}" + ratio_expr = f"{num_alias} / NULLIF({prev_denom_col}, 0)" + offset_expr = f"{self._wrap_with_fill_nulls(ratio_expr, metric)} AS {final_alias}" final_selects.append(offset_expr) # Build final query diff --git a/tests/metrics/test_advanced.py b/tests/metrics/test_advanced.py index e677b717c..9e3012399 100644 --- a/tests/metrics/test_advanced.py +++ b/tests/metrics/test_advanced.py @@ -169,6 +169,38 @@ def test_fill_nulls_with_string(): assert gadget[1] is None +def test_cumulative_fill_nulls_applies_after_window(): + sales = Model( + name="sales", + sql=""" + SELECT DATE '2024-01-01' AS day, NULL::INTEGER AS amount + UNION ALL SELECT DATE '2024-01-02', 5 + """, + primary_key="day", + dimensions=[Dimension(name="day", sql="day", type="time", granularity="day")], + metrics=[Metric(name="amount", agg="sum", sql="amount")], + ) + running_amount = Metric( + name="running_amount", + type="cumulative", + sql="sales.amount", + fill_nulls_with=0, + ) + graph = SemanticGraph() + graph.add_model(sales) + graph.add_metric(running_amount) + + sql = SQLGenerator(graph).generate( + metrics=["running_amount"], + dimensions=["sales.day"], + order_by=["sales.day"], + ) + rows = df_rows(duckdb.connect(":memory:").execute(sql)) + + assert "COALESCE(SUM(base.amount) OVER" in sql + assert [row[2] for row in rows] == [0, 5] + + def test_offset_ratio_metric(): """Test ratio metric with time offset (current / previous period).""" sales = Model( @@ -224,6 +256,37 @@ def test_offset_ratio_metric(): assert abs(rows[3][2] - 0.9) < 0.01 # Apr +def test_offset_ratio_fill_nulls_applies_to_missing_prior_period(): + sales = Model( + name="sales", + sql=""" + SELECT DATE '2024-01-01' AS month, 100 AS revenue + UNION ALL SELECT DATE '2024-02-01', 150 + """, + primary_key="month", + dimensions=[Dimension(name="month", sql="month", type="time", granularity="month")], + metrics=[Metric(name="revenue", agg="sum", sql="revenue")], + ) + growth = Metric( + name="growth", + type="ratio", + numerator="sales.revenue", + denominator="sales.revenue", + offset_window="1 month", + fill_nulls_with=0, + ) + graph = SemanticGraph() + graph.add_model(sales) + graph.add_metric(growth) + + sql = SQLGenerator(graph).generate(metrics=["growth"], dimensions=["sales.month"]) + rows = df_rows(duckdb.connect(":memory:").execute(sql)) + + assert "COALESCE(" in sql + assert rows[0][2] == 0 + assert abs(rows[1][2] - 1.5) < 0.01 + + def test_offset_ratio_metric_multi_period(): """Ratio metric with a multi-period offset_window honors the interval.""" sales = Model( @@ -311,8 +374,7 @@ def test_offset_ratio_metric_uses_base_granularity(): # Query the base time dimension WITHOUT an explicit __day suffix. sql = generator.generate(metrics=["growth"], dimensions=["sales.day"]) - # "7 days" on the dimension's day grain maps to a 7-row LAG, not a 1-row month fallback. - assert "LAG(base.revenue, 7)" in sql + assert "RANGE BETWEEN INTERVAL '7 day' PRECEDING AND INTERVAL '7 day' PRECEDING" in sql conn = duckdb.connect(":memory:") rows = df_rows(conn.execute(sql)) @@ -544,6 +606,69 @@ def test_mom_difference(): assert rows[3][2] == 60 # Apr +def test_time_comparison_fill_nulls_applies_to_missing_prior_period(): + sales = Model( + name="sales", + sql=""" + SELECT DATE '2024-01-01' AS month, 100 AS revenue + UNION ALL SELECT DATE '2024-02-01', 150 + """, + primary_key="month", + dimensions=[Dimension(name="month", sql="month", type="time", granularity="month")], + metrics=[Metric(name="revenue", agg="sum", sql="revenue")], + ) + change = Metric( + name="change", + type="time_comparison", + base_metric="sales.revenue", + comparison_type="mom", + calculation="difference", + fill_nulls_with=0, + ) + graph = SemanticGraph() + graph.add_model(sales) + graph.add_metric(change) + + sql = SQLGenerator(graph).generate(metrics=["change"], dimensions=["sales.month"]) + rows = df_rows(duckdb.connect(":memory:").execute(sql)) + + assert "COALESCE((revenue - change_prev_value), 0)" in sql + assert [row[2] for row in rows] == [0, 50] + + +def test_month_comparison_does_not_use_previous_available_sparse_row(): + sales = Model( + name="sales", + sql=""" + SELECT DATE '2024-01-01' AS sale_date, 100 AS revenue + UNION ALL SELECT DATE '2024-03-01', 180 + """, + primary_key="sale_date", + dimensions=[Dimension(name="sale_date", sql="sale_date", type="time")], + metrics=[Metric(name="revenue", agg="sum", sql="revenue")], + ) + comparison = Metric( + name="revenue_mom", + type="time_comparison", + base_metric="sales.revenue", + comparison_type="mom", + calculation="difference", + ) + graph = SemanticGraph() + graph.add_model(sales) + graph.add_metric(comparison) + + sql = SQLGenerator(graph).generate( + metrics=["revenue_mom"], + dimensions=["sales.sale_date__month"], + ) + rows = df_rows(duckdb.connect(":memory:").execute(sql)) + + assert "RANGE BETWEEN INTERVAL '1 month' PRECEDING AND INTERVAL '1 month' PRECEDING" in sql + assert rows[1][0].isoformat() == "2024-03-01" + assert rows[1][2] is None + + def test_dotted_metric_name_alias(): """Test that dotted metric names generate valid SQL aliases. diff --git a/tests/metrics/test_cohort.py b/tests/metrics/test_cohort.py index 0ceaa539e..b813b0c28 100644 --- a/tests/metrics/test_cohort.py +++ b/tests/metrics/test_cohort.py @@ -136,6 +136,42 @@ def test_cohort_with_dimension(): assert result["EU"] == 1 +def test_cohort_with_time_grain_uses_grained_output_alias(): + """Grained cohort dimensions remain addressable by their public output name.""" + events = Model( + name="events", + sql=""" + SELECT 1 AS user_id, 'web' AS platform, DATE '2024-01-01' AS ts + UNION ALL SELECT 1, 'mobile', DATE '2024-01-02' + UNION ALL SELECT 2, 'web', DATE '2024-02-01' + UNION ALL SELECT 2, 'mobile', DATE '2024-02-02' + UNION ALL SELECT 3, 'web', DATE '2024-02-03' + """, + primary_key="user_id", + dimensions=[ + Dimension(name="user_id", sql="user_id", type="categorical"), + Dimension(name="platform", sql="platform", type="categorical"), + Dimension(name="ts", sql="ts", type="time", granularity="day"), + ], + metrics=[_make_multi_platform_metric()], + ) + graph = SemanticGraph() + graph.add_model(events) + + sql = SQLGenerator(graph).generate( + metrics=["events.multi_platform_users"], + dimensions=["events.ts__month"], + order_by=["events.ts__month"], + ) + result = duckdb.connect(":memory:").execute(sql) + + assert [column[0] for column in result.description] == ["ts__month", "multi_platform_users"] + assert [(row[0].isoformat(), row[1]) for row in df_rows(result)] == [ + ("2024-01-01", 1), + ("2024-02-01", 1), + ] + + def test_cohort_outer_agg_without_sql_raises(): """Non-count outer agg without sql should raise, not emit SUM(*).""" events = _make_events_model() diff --git a/tests/metrics/test_non_additive_guard.py b/tests/metrics/test_non_additive_guard.py index 0323b8296..d24c85f57 100644 --- a/tests/metrics/test_non_additive_guard.py +++ b/tests/metrics/test_non_additive_guard.py @@ -3,10 +3,8 @@ A measure with ``non_additive_dimension`` must be aggregated over only the rows at the last (or first) value of that time dimension per group -- e.g. an account-balance snapshot summed across accounts but NOT double-counted across days. The generator -implements this by injecting a ``QUALIFY`` into the owning model's CTE for QUALIFY -dialects (DuckDB, Snowflake, BigQuery, ...). It raises ``UnsupportedMetricError`` only -for cases it does not implement: a dialect without QUALIFY, or the combination of -semi-additive handling with fan-out symmetric aggregation on the same model. +uses per-metric window markers in a nested query, so snapshot metrics and ordinary +additive metrics can share a query without filtering each other's rows. ``allow_non_additive_unsafe=True`` skips the rewrite entirely and aggregates naively (over ALL snapshots, i.e. the old, incorrect behavior). @@ -94,7 +92,8 @@ def test_semi_additive_value_is_last_snapshot(): # Semi-additive: sum of the last snapshot per account (global last-date window # collapses to the single latest snapshot when no other grouping is requested). sql = layer.compile(metrics=["accounts.balance"]) - assert "QUALIFY" in sql + assert "__sidemantic_snapshot_field" in sql + assert "OVER (" in sql # Grouped by account: last balance per account, summed -> 150 + 70 + 33 = 253. rows = layer.query( @@ -150,9 +149,9 @@ def test_grouping_by_non_additive_dimension_is_additive(): layer.add_model(_model(non_additive=True)) _seed(layer) - # No QUALIFY should be emitted -- each snapshot_date bucket stands on its own. + # No snapshot window is needed -- each snapshot_date bucket stands on its own. sql = layer.compile(metrics=["accounts.balance"], dimensions=["accounts.snapshot_date"]) - assert "QUALIFY" not in sql + assert "__sidemantic_snapshot_field" not in sql rows = layer.query( metrics=["accounts.balance"], dimensions=["accounts.snapshot_date"], order_by=["accounts.snapshot_date"] @@ -168,8 +167,8 @@ def test_escape_hatch_reverts_to_naive(): _seed(layer) sql = layer.compile(metrics=["accounts.balance"], dimensions=["accounts.account_id"]) - # No QUALIFY -> aggregates naively over all snapshots (over-counted, unsafe). - assert "QUALIFY" not in sql + # No snapshot marker -> aggregates naively over all snapshots (over-counted, unsafe). + assert "__sidemantic_snapshot_field" not in sql rows = layer.query( metrics=["accounts.balance"], dimensions=["accounts.account_id"], order_by=["accounts.account_id"] ).fetchall() @@ -182,7 +181,7 @@ def test_query_without_non_additive_metric_unaffected(): # revenue has no non_additive_dimension, so querying it alone must not raise. sql = layer.compile(metrics=["accounts.revenue"], dimensions=["accounts.region"]) assert "accounts" in sql.lower() - assert "QUALIFY" not in sql + assert "__sidemantic_snapshot_field" not in sql def test_model_without_non_additive_metric_unaffected(): @@ -190,20 +189,18 @@ def test_model_without_non_additive_metric_unaffected(): layer.add_model(_model(non_additive=False)) sql = layer.compile(metrics=["accounts.balance"], dimensions=["accounts.region"]) assert "accounts" in sql.lower() - assert "QUALIFY" not in sql + assert "__sidemantic_snapshot_field" not in sql -def test_non_qualify_dialect_raises_cleanly(): - """A dialect without QUALIFY (e.g. postgres) is rejected, not served wrong SQL.""" +def test_semi_additive_compiles_without_qualify_on_postgres(): + """The portable nested-window shape works on dialects without QUALIFY.""" layer = SemanticLayer() layer.add_model(_model(non_additive=True)) gen = SQLGenerator(layer.graph, dialect="postgres") - with pytest.raises(UnsupportedMetricError) as exc: - gen.generate(metrics=["accounts.balance"], dimensions=["accounts.region"]) - msg = str(exc.value) - assert "postgres" in msg - assert "QUALIFY" in msg - assert "allow_non_additive_unsafe" in msg + sql = gen.generate(metrics=["accounts.balance"], dimensions=["accounts.region"]) + + assert "OVER (PARTITION BY" in sql + assert "QUALIFY" not in sql def test_semi_additive_plus_fanout_symmetric_aggregate_raises(): @@ -308,9 +305,8 @@ def test_semi_additive_window_groupings_partition(monkeypatch): assert rows == {"east": 110, "west": 210} -def test_conflicting_semi_additive_metrics_raise(): +def test_opening_and_closing_snapshot_metrics_compose(): from sidemantic import Dimension, Metric, Model, SemanticLayer - from sidemantic.core.semantic_layer import UnsupportedMetricError layer = SemanticLayer() con = layer.adapter.conn @@ -332,12 +328,16 @@ def test_conflicting_semi_additive_metrics_raise(): ], ) ) - with pytest.raises(UnsupportedMetricError, match="conflicting"): - layer.compile(metrics=["bal.closing", "bal.opening"]) + sql = layer.compile(metrics=["bal.closing", "bal.opening"]) + rows = layer.query(metrics=["bal.closing", "bal.opening"]).fetchall() + + assert "MAX(" in sql + assert "MIN(" in sql + assert rows == [(110, 100)] def test_graph_metric_wrapping_semi_additive_measure_is_planned(): - """PR review: a graph metric wrapping a non_additive measure must still emit the QUALIFY.""" + """A graph metric wrapping a non-additive measure retains its snapshot plan.""" from sidemantic import Dimension, Metric, Model, SemanticLayer layer = SemanticLayer() @@ -362,5 +362,20 @@ def test_graph_metric_wrapping_semi_additive_measure_is_planned(): ) layer.add_metric(Metric(name="wrapped_balance", sql="bal.total_balance")) sql = layer.compile(metrics=["wrapped_balance"], dimensions=["bal.account"]) - assert "QUALIFY" in sql + assert "__sidemantic_snapshot_field" in sql assert dict(layer.query(metrics=["wrapped_balance"], dimensions=["bal.account"]).fetchall()) == {"A": 110, "B": 210} + + +def test_semi_additive_and_additive_metrics_keep_independent_row_sets(): + """A snapshot metric must not remove rows from additive sibling metrics.""" + layer = SemanticLayer() + layer.add_model(_model(non_additive=True)) + _seed(layer) + + rows = layer.query( + metrics=["accounts.balance", "accounts.revenue"], + dimensions=["accounts.account_id"], + order_by=["accounts.account_id"], + ).fetchall() + + assert rows == [("A", 150.0, 2.0), ("B", 70.0, 2.0), ("C", 33.0, 2.0)] From d08bdbb9dec70f5447b2f3b1fe4cb13d9b955ff8 Mon Sep 17 00:00:00 2001 From: Nico Ritschel Date: Sat, 1 Aug 2026 22:28:46 -0700 Subject: [PATCH 5/5] Apply rollup fallback to serving surfaces --- sidemantic/api_server.py | 107 +++++++++++++++++++++++++++----- sidemantic/mcp_server.py | 23 ++++++- sidemantic/workbench/app.py | 30 +++++++-- tests/server/test_api_server.py | 95 ++++++++++++++++++++++++++++ tests/test_mcp_server.py | 43 ++++++++++++- 5 files changed, 272 insertions(+), 26 deletions(-) diff --git a/sidemantic/api_server.py b/sidemantic/api_server.py index 0237b81cc..f33e95a1d 100644 --- a/sidemantic/api_server.py +++ b/sidemantic/api_server.py @@ -104,6 +104,7 @@ class StructuredQueryRequest(BaseModel): ungrouped: bool = False parameters: dict[str, Any] | None = None use_preaggregations: bool | None = None + preagg_strict: bool | None = None timezone: str | None = None def resolved_filters(self) -> list[str]: @@ -333,7 +334,7 @@ def require_auth( async def handle_value_error(_request: Request, exc: ValueError): return JSONResponse({"error": str(exc)}, status_code=400) - from sidemantic.core.semantic_layer import SecurityError + from sidemantic.core.semantic_layer import PreaggregationStrictError, SecurityError @app.exception_handler(SecurityError) async def handle_security_error(_request: Request, exc: SecurityError): @@ -341,6 +342,10 @@ async def handle_security_error(_request: Request, exc: SecurityError): # access gate denied the request). Map to 403 Forbidden. return JSONResponse({"error": str(exc)}, status_code=403) + @app.exception_handler(PreaggregationStrictError) + async def handle_preagg_strict_error(_request: Request, exc: PreaggregationStrictError): + return JSONResponse({"error": str(exc)}, status_code=409) + def resolve_user_attributes(request: Request) -> dict | None: """Parse per-request user attributes from the trusted user header. @@ -565,22 +570,40 @@ def run_query( filters = payload.resolved_filters() for filter_str in filters: validate_filter_expression(filter_str, dialect=current_layer.dialect) - sql = current_layer.compile( - dimensions=payload.dimensions, - metrics=payload.metrics, - filters=filters, - segments=payload.segments or None, - order_by=payload.order_by or None, - limit=payload.limit, - offset=payload.offset, - ungrouped=payload.ungrouped, - parameters=payload.parameters, - use_preaggregations=payload.use_preaggregations, + + def compile_query(use_preaggregations: bool | None) -> str: + return current_layer.compile( + dimensions=payload.dimensions, + metrics=payload.metrics, + filters=filters, + segments=payload.segments or None, + order_by=payload.order_by or None, + limit=payload.limit, + offset=payload.offset, + ungrouped=payload.ungrouped, + parameters=payload.parameters, + use_preaggregations=use_preaggregations, + user_attributes=user_attributes, + timezone=payload.timezone, + ) + + sql = compile_query(payload.use_preaggregations) + use_preaggs = ( + payload.use_preaggregations + if payload.use_preaggregations is not None + else current_layer.use_preaggregations + ) + strict = payload.preagg_strict if payload.preagg_strict is not None else current_layer.preagg_strict + table, executed_sql = _query_table_with_preagg_fallback( + app, + current_layer, + sql, + lambda: compile_query(False), + use_preaggs=use_preaggs, + strict=strict, user_attributes=user_attributes, - timezone=payload.timezone, ) - table = _query_table(app, current_layer, sql, user_attributes=user_attributes) - return _build_query_response(request, current_layer, table, sql=sql, format_override=format) + return _build_query_response(request, current_layer, table, sql=executed_sql, format_override=format) @app.post("/sql/compile", dependencies=[Depends(require_auth)]) def compile_sql(payload: SQLRequest, request: Request) -> dict[str, str]: @@ -617,12 +640,26 @@ def run_sql( user_attributes=user_attributes, transport="HTTP /sql", ) - table = _query_table(app, current_layer, rewritten_sql, user_attributes=user_attributes) + table, executed_sql = _query_table_with_preagg_fallback( + app, + current_layer, + rewritten_sql, + lambda: rewrite_transport_sql( + current_layer, + query, + user_attributes=user_attributes, + transport="HTTP /sql", + use_preaggregations=False, + ), + use_preaggs=current_layer.use_preaggregations, + strict=current_layer.preagg_strict, + user_attributes=user_attributes, + ) return _build_query_response( request, current_layer, table, - sql=rewritten_sql, + sql=executed_sql, original_sql=query, format_override=format, ) @@ -695,6 +732,42 @@ def _execute_to_table(layer: SemanticLayer, sql: str) -> Any: return record_batch_reader_to_table(reader) +def _query_table_with_preagg_fallback( + app: FastAPI, + layer: SemanticLayer, + sql: str, + recompile_raw, + *, + use_preaggs: bool, + strict: bool, + user_attributes: dict | None = None, +) -> tuple[Any, str]: + """Execute routed SQL, falling back to raw tables when its rollup is missing.""" + from sidemantic.core.semantic_layer import PreaggregationStrictError + + if not use_preaggs: + return _query_table(app, layer, sql, user_attributes=user_attributes), sql + + used_preagg = "used_preagg=true" in sql + if strict and not used_preagg: + raise PreaggregationStrictError( + "Strict pre-aggregation mode: no pre-aggregation matched this query " + "(its metrics/dimensions/granularity are not covered by any rollup)." + ) + try: + return _query_table(app, layer, sql, user_attributes=user_attributes), sql + except Exception as exc: + if not used_preagg or not layer._is_missing_relation_error(exc): + raise + if strict: + raise PreaggregationStrictError( + "Strict pre-aggregation mode: the matching pre-aggregation table is not built. " + "Materialize it (e.g. `sidemantic preagg refresh`) before querying." + ) from exc + raw_sql = recompile_raw() + return _query_table(app, layer, raw_sql, user_attributes=user_attributes), raw_sql + + def _query_table(app: FastAPI, layer: SemanticLayer, sql: str, user_attributes: dict | None = None) -> Any: """Return the Arrow table for ``sql``, served from the result cache if enabled. diff --git a/sidemantic/mcp_server.py b/sidemantic/mcp_server.py index e1e35f88e..a5356d3ca 100644 --- a/sidemantic/mcp_server.py +++ b/sidemantic/mcp_server.py @@ -448,8 +448,27 @@ def run_query( if dry_run: return {"sql": sql} - # Execute query via adapter (works with all database backends) - result = layer.adapter.execute(sql) + def recompile_raw(): + return layer.compile( + dimensions=dimensions or [], + metrics=metrics or [], + filters=[where] if where else None, + segments=segments, + order_by=order_by, + limit=limit or None, + offset=offset or None, + ungrouped=ungrouped, + use_preaggregations=False, + user_attributes=get_user_attributes(), + ) + + result = layer._execute_with_preagg_fallback( + sql, + recompile_raw, + use_preaggs=layer.use_preaggregations, + strict=layer.preagg_strict, + used_preagg="used_preagg=true" in sql, + ) # Convert to list of dicts with JSON-compatible values rows = result.fetchall() diff --git a/sidemantic/workbench/app.py b/sidemantic/workbench/app.py index 06c79cbaa..595bde6f2 100644 --- a/sidemantic/workbench/app.py +++ b/sidemantic/workbench/app.py @@ -579,17 +579,35 @@ def action_run_query(self) -> None: if not sql: return - # Execute query and get rendered SQL - from sidemantic.sql.query_rewriter import QueryRewriter + # Route semantic SQL through the shared security-aware rewriter. + from sidemantic.core.transport_security import rewrite_transport_sql - rewriter = QueryRewriter(self.layer.graph, dialect=self.layer.dialect) - rendered_sql = rewriter.rewrite(sql) + rendered_sql = rewrite_transport_sql( + self.layer, + sql, + user_attributes=None, + transport="Workbench", + ) # Store rendered SQL self.last_rendered_sql = rendered_sql - # Execute the query - result = self.layer.adapter.execute(rendered_sql) + def recompile_raw(): + return rewrite_transport_sql( + self.layer, + sql, + user_attributes=None, + transport="Workbench", + use_preaggregations=False, + ) + + result = self.layer._execute_with_preagg_fallback( + rendered_sql, + recompile_raw, + use_preaggs=self.layer.use_preaggregations, + strict=self.layer.preagg_strict, + used_preagg="used_preagg=true" in rendered_sql, + ) # Get column names and rows columns = [desc[0] for desc in result.description] diff --git a/tests/server/test_api_server.py b/tests/server/test_api_server.py index 8aa4c1d74..ba4fb4972 100644 --- a/tests/server/test_api_server.py +++ b/tests/server/test_api_server.py @@ -640,3 +640,98 @@ def test_json_responses_use_arrow_reader_for_generic_adapters(): assert response.status_code == 200 assert response.json()["rows"] == [{"order_count": 7}] + + +def _build_unbuilt_rollup_client(tmp_path: Path, preagg_strict: bool = False) -> TestClient: + """Build a client whose matching pre-aggregation was never materialized.""" + from sidemantic.core.pre_aggregation import PreAggregation + + db_path = tmp_path / "preagg-warehouse.duckdb" + conn = duckdb.connect(str(db_path)) + conn.execute("create table orders (id integer, status varchar, amount double)") + conn.executemany( + "insert into orders values (?, ?, ?)", + [(1, "completed", 10.0), (2, "completed", 20.0), (3, "pending", 5.0)], + ) + conn.close() + + layer = SemanticLayer( + connection=f"duckdb:///{db_path}", + auto_register=False, + use_preaggregations=True, + preagg_strict=preagg_strict, + ) + layer.add_model( + Model( + name="orders", + table="orders", + primary_key="id", + dimensions=[Dimension(name="status", sql="status", type="categorical")], + metrics=[Metric(name="revenue", agg="sum", sql="amount")], + pre_aggregations=[PreAggregation(name="by_status", measures=["revenue"], dimensions=["status"])], + ) + ) + return TestClient(create_app(layer)) + + +def test_query_falls_back_to_raw_when_rollup_missing(tmp_path): + client = _build_unbuilt_rollup_client(tmp_path) + + response = client.post("/query", json={"metrics": ["orders.revenue"], "dimensions": ["orders.status"]}) + + assert response.status_code == 200 + rows = sorted((row["status"], row["revenue"]) for row in response.json()["rows"]) + assert rows == [("completed", 30.0), ("pending", 5.0)] + assert "used_preagg=true" not in response.json()["sql"] + + +def test_query_strict_mode_returns_409_when_rollup_missing(tmp_path): + client = _build_unbuilt_rollup_client(tmp_path, preagg_strict=True) + + response = client.post("/query", json={"metrics": ["orders.revenue"], "dimensions": ["orders.status"]}) + + assert response.status_code == 409 + assert "not built" in response.json()["error"] + + +def test_query_strict_override_via_payload(tmp_path): + client = _build_unbuilt_rollup_client(tmp_path) + + response = client.post( + "/query", + json={"metrics": ["orders.revenue"], "dimensions": ["orders.status"], "preagg_strict": True}, + ) + + assert response.status_code == 409 + + +def test_sql_endpoint_falls_back_to_raw_when_rollup_missing(tmp_path): + client = _build_unbuilt_rollup_client(tmp_path) + + compiled = client.post( + "/sql/compile", + json={"query": "SELECT orders.revenue, orders.status FROM orders"}, + ) + response = client.post( + "/sql", + json={"query": "SELECT orders.revenue, orders.status FROM orders"}, + ) + + assert compiled.status_code == 200 + assert "used_preagg=true" in compiled.json()["sql"] + assert response.status_code == 200 + rows = sorted((row["status"], row["revenue"]) for row in response.json()["rows"]) + assert rows == [("completed", 30.0), ("pending", 5.0)] + assert "used_preagg=true" not in response.json()["sql"] + + +def test_sql_endpoint_strict_mode_returns_409_when_rollup_missing(tmp_path): + client = _build_unbuilt_rollup_client(tmp_path, preagg_strict=True) + + response = client.post( + "/sql", + json={"query": "SELECT orders.revenue, orders.status FROM orders"}, + ) + + assert response.status_code == 409 + assert "not built" in response.json()["error"] diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 37130e4e3..6bdb74959 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -14,8 +14,10 @@ ensure_fake_mcp() -from sidemantic import Metric, Model +from sidemantic import Dimension, Metric, Model, SemanticLayer +from sidemantic.core.pre_aggregation import PreAggregation from sidemantic.core.relationship import Relationship +from sidemantic.core.semantic_layer import PreaggregationStrictError from sidemantic.mcp_server import ( _convert_to_json_compatible, _format_join_condition, @@ -261,6 +263,45 @@ def test_run_query_metrics_only(demo_layer): assert "COUNT" in result["sql"].upper() +def _unbuilt_rollup_layer(*, strict: bool = False) -> SemanticLayer: + layer = SemanticLayer(auto_register=False, use_preaggregations=True, preagg_strict=strict) + layer.adapter.execute("create table orders (id integer, status varchar, amount double)") + layer.adapter.execute( + "insert into orders values (1, 'completed', 10.0), (2, 'completed', 20.0), (3, 'pending', 5.0)" + ) + layer.add_model( + Model( + name="orders", + table="orders", + primary_key="id", + dimensions=[Dimension(name="status", sql="status", type="categorical")], + metrics=[Metric(name="revenue", agg="sum", sql="amount")], + pre_aggregations=[PreAggregation(name="by_status", measures=["revenue"], dimensions=["status"])], + ) + ) + return layer + + +def test_run_query_falls_back_when_rollup_is_missing(monkeypatch): + import sidemantic.mcp_server as mcp_server + + monkeypatch.setattr(mcp_server, "_layer", _unbuilt_rollup_layer()) + + result = run_query(metrics=["orders.revenue"], dimensions=["orders.status"]) + + rows = sorted((row["status"], row["revenue"]) for row in result["rows"]) + assert rows == [("completed", 30.0), ("pending", 5.0)] + + +def test_run_query_strict_mode_rejects_missing_rollup(monkeypatch): + import sidemantic.mcp_server as mcp_server + + monkeypatch.setattr(mcp_server, "_layer", _unbuilt_rollup_layer(strict=True)) + + with pytest.raises(PreaggregationStrictError, match="not built"): + run_query(metrics=["orders.revenue"], dimensions=["orders.status"]) + + def test_run_query_decimal_conversion(demo_layer): """Test that Decimal values are converted to float for JSON serialization.""" result = run_query(