Apply rollup fallback across serving surfaces

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.
 

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

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]

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"]

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(