Enforce model required filters across query clients

docs/native-format.md

@@ -128,6 +128,8 @@ At least one of `table`, `sql`, or `source_uri` should be present unless the mod
 | `metrics` | No | List of model-local metrics. |
 | `relationships` | No | List of relationships from this model to other models. |
 | `segments` | No | List of model-local named filters. |
+| `required_filters` | No | Fields that every query using the model must filter. Enforced by Python-hosted clients before Python or Rust compilation. |
+| `required_filters_or_groups` | No | Fields that every query using the model must filter or group by, independently for each listed field. |
 | `pre_aggregations` | No | List of pre-aggregation definitions. |
 | `default_time_dimension` | No | Time dimension to add by default when the query needs time grouping. |
 | `default_grain` | No | Default time grain for the default time dimension. |
@@ -140,6 +142,63 @@ also accept compatibility input aliases: model-level `measures` for `metrics`,
 dimension/metric `expr` for `sql`, and metric `measure` for `sql`. Exports use
 canonical field names.
 
+### Required filters
+
+Use model-level caller obligations to reject queries that omit a date range or
+combine rows across a required grouping dimension:
+
+```yaml
+models:
+  - name: daily_activity
+    table: warehouse.daily_activity
+    required_filters: [event_date]
+    required_filters_or_groups: [test_name]
+    dimensions:
+      - name: event_date
+        type: time
+        granularity: day
+      - name: test_name
+        type: categorical
+    metrics:
+      - name: activity_count
+        agg: count
+```
+
+```sh
+sidemantic query --models ./models --metric daily_activity.activity_count \
+  --dimension daily_activity.test_name \
+  --filter "daily_activity.event_date >= '2026-01-01'"
+```
+
+Each listed field is mandatory, including on models reached through metric
+dependencies, filters, and intermediate joins. Declarations accept local names
+or names qualified with the declaring model. Inheritance accumulates parent and
+child requirements, rebinding inherited fields to the child. Empty lists are the
+default and impose no obligations.
+
+Matching uses SQL column references, not text or string literals. A time-grained
+reference such as `daily_activity.event_date__month` matches `event_date`.
+`required_filters` cannot be satisfied by grouping; `required_filters_or_groups`
+accepts either a filter or grouping for **each** field. Structured dimensions count
+as grouping unless `ungrouped` is set. SQL clients must use an explicit `GROUP BY`
+(field or selected alias). Filters on an outer query do not satisfy obligations
+inside a CTE or subquery. The check requires a field reference; it does not prove
+selectivity or impose a maximum date range.
+
+Selected segments and resolved Explore filters can satisfy the obligation.
+Explore `default_filters` still apply only when filters are omitted; replacing
+them with an unrelated filter fails the model requirement. Explore mandatory
+filters still always apply. Model `invariant_filters`, security policies, and
+metric-internal filters do not substitute for the caller obligation.
+
+The Python CLI, HTTP `/query`, MCP structured queries, and standard semantic SQL
+(including PostgreSQL wire queries) enforce these rules before engine dispatch.
+Missing requirements raise `ValueError` and HTTP 400. Raw SQL bypass endpoints are
+disabled while requirements are declared. Yardstick SQL is rejected for these
+graphs; use structured queries or standard semantic SQL. Standalone Rust and
+DuckDB extension runtimes do not yet support these model options; exported
+definitions retain them and unsupported native loaders reject them.
+
 Single-column primary key:
 
 ```yaml

sidemantic-schema.json

@@ -5453,6 +5453,22 @@
             "title": "Relationships",
             "type": "array"
           },
+          "required_filters": {
+            "description": "Fields callers must filter",
+            "items": {
+              "type": "string"
+            },
+            "title": "Required Filters",
+            "type": "array"
+          },
+          "required_filters_or_groups": {
+            "description": "Fields callers must filter or group by, independently for each field",
+            "items": {
+              "type": "string"
+            },
+            "title": "Required Filters Or Groups",
+            "type": "array"
+          },
           "schema_exposure": {
             "anyOf": [
               {

sidemantic/adapters/sidemantic.py

@@ -63,6 +63,8 @@
     "relationships",
     "segments",
     "invariant_filters",
+    "required_filters",
+    "required_filters_or_groups",
     "pre_aggregations",
     "security",
     "default_time_dimension",
@@ -778,6 +780,8 @@ def _parse_model(self, model_def: dict, *, source_path: Path | None = None) -> M
             "auto_dimensions",
             "meta",
             "invariant_filters",
+            "required_filters",
+            "required_filters_or_groups",
         ]:
             if field in model_def:
                 model_kwargs[field] = model_def.get(field)
@@ -1002,6 +1006,9 @@ def _export_model(self, model: Model) -> dict:
             result["meta"] = model.meta
         if model.invariant_filters:
             result["invariant_filters"] = model.invariant_filters
+        for policy in ("required_filters", "required_filters_or_groups"):
+            if getattr(model, policy):
+                result[policy] = getattr(model, policy)
         if model.auto_dimensions:
             result["auto_dimensions"] = True
         if model.schema_exposure is not None:

sidemantic/core/inheritance.py

@@ -41,6 +41,8 @@ def merge_model(child: Model, parent: Model) -> Model:
         *merged_data.get("invariant_filters", []),
         *child_data.get("invariant_filters", []),
     ]
+    for policy in ("required_filters", "required_filters_or_groups"):
+        merged_data[policy] = list(dict.fromkeys([*merged_data.get(policy, []), *child_data.get(policy, [])]))
 
     # Merge lists (dimensions, metrics, relationships, segments)
     # Child's items are added to parent's items

sidemantic/core/model.py

@@ -2,7 +2,7 @@
 
 from typing import Any, Literal
 
-from pydantic import Field
+from pydantic import Field, model_validator
 
 from sidemantic.core.dimension import Dimension
 from sidemantic.core.governance import GovernedObject
@@ -51,6 +51,11 @@ class Model(GovernedObject):
         default_factory=list,
         description="SQL predicates that always constrain this model before joins and aggregation",
     )
+    required_filters: list[str] = Field(default_factory=list, description="Fields callers must filter")
+    required_filters_or_groups: list[str] = Field(
+        default_factory=list, description="Fields callers must filter or group by, independently for each field"
+    )
+
     pre_aggregations: list[PreAggregation] = Field(
         default_factory=list, description="Pre-aggregation definitions for query optimization"
     )
@@ -81,6 +86,23 @@ class Model(GovernedObject):
         None, description="Model-level access control and row-level security policy"
     )
 
+    @model_validator(mode="after")
+    def normalize_required_fields(self) -> "Model":
+        """Store local field names so inherited obligations bind to the child model."""
+        import re
+
+        for policy in ("required_filters", "required_filters_or_groups"):
+            normalized = []
+            for reference in getattr(self, policy):
+                field = reference.removeprefix(f"{self.name}.")
+                if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", field):
+                    raise ValueError(f"{policy} must name fields on {self.name}: {reference!r}")
+                normalized.append(field)
+            normalized = list(dict.fromkeys(normalized))
+            if normalized != getattr(self, policy):
+                setattr(self, policy, normalized)
+        return self
+
     def __init__(self, **data):
         super().__init__(**data)
 

sidemantic/core/required_filters.py

@@ -0,0 +1,157 @@
+"""Caller filter obligations shared by structured and SQL query entrypoints."""
+
+from sqlglot import exp, parse_one
+from sqlglot.errors import SqlglotError
+from sqlglot.optimizer.scope import traverse_scope
+
+
+def has_required_filters(graph) -> bool:
+    return any(model.required_filters or model.required_filters_or_groups for model in graph.models.values())
+
+
+def _parse(expression: str, dialect: str) -> exp.Expression:
+    try:
+        return parse_one(expression, dialect=dialect)
+    except SqlglotError as error:
+        raise ValueError(f"Cannot validate required filters: invalid SQL expression {expression!r}") from error
+
+
+def _field(graph, model_name: str, name: str) -> str:
+    model = graph.models.get(model_name)
+    # Only strip a grain when it actually identifies a declared time dimension.
+    if model and "__" in name and not model.get_dimension(name):
+        base = name.rsplit("__", 1)[0]
+        dimension = model.get_dimension(base)
+        if dimension and dimension.type == "time":
+            name = base
+    return f"{model_name}.{name}"
+
+
+def _references(graph, expression, models, aliases=None) -> set[str]:
+    """Bind outer columns only; subquery columns belong to a separate query."""
+    references = set()
+    aliases = aliases or {}
+    for node in expression.walk(prune=lambda node: isinstance(node, (exp.Select, exp.Subquery))):
+        if not isinstance(node, exp.Column):
+            continue
+        owner = aliases.get(node.table, node.table)
+        if owner.endswith("_cte") and owner not in graph.models:
+            owner = owner.removesuffix("_cte")
+        if not owner:
+            candidates = [
+                name
+                for name in models
+                if name in graph.models
+                and (
+                    graph.models[name].get_dimension(_field(graph, name, node.name).split(".", 1)[1])
+                    or graph.models[name].get_metric(node.name)
+                )
+            ]
+            if len(candidates) != 1:
+                continue
+            owner = candidates[0]
+        if owner in graph.models:
+            references.add(_field(graph, owner, node.name))
+    return references
+
+
+def _enforce(graph, models, filtered, grouped):
+    for name in sorted(models):
+        model = graph.models.get(name)
+        if model is None:
+            continue
+        for field in model.required_filters:
+            reference = _field(graph, name, field)
+            if reference not in filtered:
+                raise ValueError(f"a query on {name} must filter {reference}")
+        for field in model.required_filters_or_groups:
+            reference = _field(graph, name, field)
+            if reference not in filtered | grouped:
+                raise ValueError(f"a query on {name} must filter or group by {reference}")
+
+
+def validate_required_filters(
+    graph,
+    *,
+    metrics=(),
+    dimensions=(),
+    filters=(),
+    segments=(),
+    ungrouped=False,
+    base_model=None,
+    parameters=None,
+    dialect="duckdb",
+):
+    """Check resolved caller filters before selecting a compiler engine."""
+    if not has_required_filters(graph):
+        return
+    from sidemantic.core.parameter import ParameterSet
+    from sidemantic.sql.generator import SQLGenerator
+
+    generator = SQLGenerator(graph, dialect=dialect, base_model=base_model)
+    parameter_set = ParameterSet(graph.parameters, parameters, dialect=dialect)
+    predicates = [parameter_set.interpolate(predicate) for predicate in filters or []]
+    predicates += generator._resolve_segments(segments or [], interpolate=parameter_set.interpolate)
+    models = generator._participating_models(
+        generator._find_required_models(list(metrics or []), list(dimensions or []), predicates)
+    )
+    filtered = set()
+    for predicate in predicates:
+        filtered.update(_references(graph, _parse(predicate, dialect), models))
+    grouped = set()
+    if not ungrouped:
+        for reference in dimensions or []:
+            grouped.update(_references(graph, _parse(reference, dialect), models))
+    _enforce(graph, models, filtered, grouped)
+
+
+def validate_required_filters_sql(graph, sql, *, dialect="duckdb"):
+    """Check each SQL read scope independently, before Rust or Python rewriting.
+
+    WHERE predicates and explicit GROUP BY expressions satisfy obligations. Outer
+    filters on an already aggregated subquery do not constrain its source read.
+    """
+    if not has_required_filters(graph):
+        return
+    from sidemantic.sql.generator import SQLGenerator
+
+    parsed = _parse(sql, dialect)
+    for table in parsed.find_all(exp.Table):
+        if isinstance(table.this, exp.Anonymous) and table.this.name.lower() == "yardstick":
+            raise ValueError("Required model filters require standard semantic SQL; yardstick() is unsupported")
+    for scope in traverse_scope(parsed):
+        select = scope.expression
+        if not isinstance(select, exp.Select):
+            continue
+        aliases = {
+            alias: source.name
+            for alias, source in scope.sources.items()
+            if isinstance(source, exp.Table) and source.name in graph.models
+        }
+        models = list(dict.fromkeys(aliases.values()))
+        references = set()
+        for column in scope.columns:
+            references.update(_references(graph, column, models, aliases))
+        metrics = [ref for ref in references if graph.models[ref.split(".")[0]].get_metric(ref.split(".")[1])]
+        for column in scope.columns:
+            reference = f"{column.table}.{column.name}" if column.table else column.name
+            if reference in graph.metrics:
+                metrics.append(reference)
+        dimensions = [ref for ref in references if ref not in metrics]
+        generator = SQLGenerator(graph, dialect=dialect)
+        models = generator._participating_models([*models, *generator._find_required_models(metrics, dimensions)])
+        where = select.args.get("where")
+        filtered = _references(graph, where.this, models, aliases) if where else set()
+        grouped = set()
+        group = select.args.get("group")
+        if group:
+            selected_aliases = {item.alias: item.this for item in select.expressions if isinstance(item, exp.Alias)}
+            for expression in group.expressions:
+                if isinstance(expression, exp.Literal) and expression.is_int:
+                    index = int(expression.this) - 1
+                    if 0 <= index < len(select.expressions):
+                        expression = select.expressions[index]
+                elif isinstance(expression, exp.Column) and not expression.table:
+                    expression = selected_aliases.get(expression.name, expression)
+                grouped.update(_references(graph, expression, models, aliases))
+        _enforce(graph, models, filtered, grouped)

sidemantic/core/semantic_layer.py

@@ -1229,6 +1229,20 @@ def compile(
         metrics = metrics or []
         dimensions = dimensions or []
 
+        from sidemantic.core.required_filters import validate_required_filters
+
+        validate_required_filters(
+            self.graph,
+            metrics=metrics,
+            dimensions=dimensions,
+            filters=filters,
+            segments=segments,
+            ungrouped=ungrouped,
+            base_model=consumption_base_model,
+            parameters=parameters,
+            dialect=dialect or self.dialect,
+        )
+
         if with_totals and (limit is not None or offset is not None):
             raise ValueError(
                 "with_totals cannot be combined with limit/offset: the grand-total row shares "
@@ -2287,6 +2301,12 @@ def sql(self, query: str, *, user_attributes: dict | None = None):
         use_cache = user_attributes is None and not yardstick_warnings_may_apply(self.graph, query)
         cache_key = (
             getattr(self.graph, "_version", 0),
+            # Policy lists can be assigned or mutated without a graph version
+            # change. A cached rewrite must still satisfy the current obligations.
+            tuple(
+                (name, tuple(model.required_filters), tuple(model.required_filters_or_groups))
+                for name, model in self.graph.models.items()
+            ),
             self.dialect,
             self.use_preaggregations,
             self.enforce_visibility,