Make semantic planning deterministic and faster

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

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

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

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):

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

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

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

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

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

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

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

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

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,14 +334,18 @@ 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):
         # A secured model was queried without sufficient user attributes (or an
         # 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.