refactor(api): build the model pages on climate-ref's filters

backend/src/ref_backend/core/mip_eras.py

@@ -3,12 +3,11 @@
 from collections.abc import Collection
 from typing import Any, Literal
 
-from sqlalchemy import ColumnElement, select
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from climate_ref import models
 from climate_ref.models.execution import ExecutionGroupSourceType
-from climate_ref.query.filters import ExecutionScope
 from climate_ref.results import InputDatasetFilter
 from climate_ref_core.datasets import SourceDatasetType
 
@@ -59,11 +58,6 @@ def eras_in_scope(mip_era: MipEra | None) -> tuple[SourceDatasetType, ...]:
     return source_types_for(mip_era) or CMIP_ERAS
 
 
-def era_group_clauses(mip_era: MipEra) -> list[ColumnElement[bool]]:
-    """Clauses over `ExecutionGroup` keeping the groups that draw on `mip_era`, or on no model data."""
-    return ExecutionScope(**era_scope(mip_era)).group_clauses()
-
-
 def source_id_datasets(source_id: str | None) -> InputDatasetFilter | None:
     """Match executions that ran against a model, which selectors do not always record."""
     return InputDatasetFilter(facets={"source_id": source_id}) if source_id else None

backend/src/ref_backend/core/model_ensemble.py

@@ -5,12 +5,13 @@
 from statistics import fmean, pstdev, quantiles
 from typing import Any, NamedTuple
 
-from sqlalchemy import select
+from sqlalchemy import Select, select
 from sqlalchemy.orm import Session
 
 from climate_ref import models
-from climate_ref.query import select_latest_executions
-from ref_backend.core.mip_eras import MipEra, cv_column, era_group_clauses
+from climate_ref.query import select_scalar_values
+from climate_ref.results import MetricValueFilter
+from ref_backend.core.mip_eras import MipEra, cv_column, era_scope
 from ref_backend.models.model_runs import EnsembleComparison, EnsembleStatistics
 
 #: Dimensions that say which run produced a value rather than what the value measures.
@@ -81,6 +82,12 @@ class ScalarRow(NamedTuple):
     dimensions: tuple[str | None, ...]
 
 
+def _model_values(mip_era: MipEra | None, **kwargs: Any) -> Select[Any]:
+    """Select each group's latest model values, in the era's groups and those with no model input."""
+    filters = MetricValueFilter(latest_only=True, **era_scope(mip_era), **kwargs)
+    return select_scalar_values(filters).order_by(None)
+
+
 def _scalar_rows(
     session: Session,
     *,
@@ -90,51 +97,38 @@ def _scalar_rows(
 ) -> list[ScalarRow]:
     value = models.ScalarMetricValue
     columns = [cv_column(value, name) for name in dimensions]
-    # A retried group holds a value per attempt, so only the unretracted execution that decided it counts.
-    latest = select_latest_executions().subquery()
-
     statement = (
-        select(
+        _model_values(mip_era, dimensions={"kind": "model"})
+        .with_only_columns(
             models.ExecutionGroup.diagnostic_id,
             cv_column(value, "source_id"),
             value.value,
             value.attributes,
             *columns,
+            maintain_column_froms=True,
         )
-        .join(models.Execution, value.execution_id == models.Execution.id)
-        .join(models.ExecutionGroup, models.Execution.execution_group_id == models.ExecutionGroup.id)
-        .join(models.Diagnostic, models.ExecutionGroup.diagnostic_id == models.Diagnostic.id)
-        .join(latest, latest.c.execution_id == models.Execution.id)
         .where(
-            models.ExecutionGroup.diagnostic_version == models.Diagnostic.promoted_version,
             models.ExecutionGroup.diagnostic_id.in_(diagnostic_ids),
-            models.Execution.retracted.is_(False),
-            cv_column(value, "kind") == "model",
             cv_column(value, "source_id").isnot(None),
             value.value.isnot(None),
         )
     )
-    if mip_era:
-        statement = statement.where(*era_group_clauses(mip_era))
 
     return [
         ScalarRow(row[0], row[1], row[2], _units(row[3]), tuple(row[4:]))
         for row in session.execute(statement).all()
     ]
 
 
-def _diagnostics_for_model(session: Session, *, source_id: str) -> list[int]:
+def _diagnostics_for_model(
+    session: Session, *, source_id: str, mip_era: MipEra | None, diagnostic_slug: str | None
+) -> list[int]:
     """Find the diagnostics that recorded a scalar value for this model."""
     statement = (
-        select(models.ExecutionGroup.diagnostic_id)
-        .join(models.Execution, models.Execution.execution_group_id == models.ExecutionGroup.id)
-        .join(models.ScalarMetricValue, models.ScalarMetricValue.execution_id == models.Execution.id)
-        .join(models.Diagnostic, models.ExecutionGroup.diagnostic_id == models.Diagnostic.id)
-        .where(
-            models.ExecutionGroup.diagnostic_version == models.Diagnostic.promoted_version,
-            cv_column(models.ScalarMetricValue, "source_id") == source_id,
-            cv_column(models.ScalarMetricValue, "kind") == "model",
+        _model_values(
+            mip_era, diagnostic_slug=diagnostic_slug, dimensions={"kind": "model", "source_id": source_id}
         )
+        .with_only_columns(models.ExecutionGroup.diagnostic_id, maintain_column_froms=True)
         .distinct()
     )
     return list(session.scalars(statement))
@@ -156,17 +150,15 @@ def ensemble_comparisons(
     """
     dimensions = grouping_dimensions()
 
-    model_diagnostics = _diagnostics_for_model(session, source_id=source_id)
+    model_diagnostics = _diagnostics_for_model(
+        session, source_id=source_id, mip_era=mip_era, diagnostic_slug=diagnostic_slug
+    )
     diagnostics = {
         diagnostic.id: diagnostic
         for diagnostic in session.scalars(
             select(models.Diagnostic).where(models.Diagnostic.id.in_(model_diagnostics))
         )
     }
-    if diagnostic_slug:
-        diagnostics = {
-            key: diagnostic for key, diagnostic in diagnostics.items() if diagnostic.slug == diagnostic_slug
-        }
     if not diagnostics:
         return []
 

backend/src/ref_backend/core/model_runs.py

@@ -4,12 +4,22 @@
 from collections.abc import Sequence
 from typing import Any, NamedTuple
 
-from sqlalchemy import CompoundSelect, Select, case, distinct, func, select, union
-from sqlalchemy.orm import Session, aliased
+from sqlalchemy import Select, Subquery, case, distinct, func, select, union
+from sqlalchemy.orm import Session
 
 from climate_ref import models
-from climate_ref.query import select_latest_executions
-from ref_backend.core.mip_eras import MipEra, cv_column, dataset_model_for, eras_in_scope, mip_era_for
+from climate_ref.models.execution import execution_datasets
+from climate_ref.query import select_execution_group_ids, select_execution_groups
+from climate_ref.results import ExecutionGroupFilter
+from ref_backend.core.mip_eras import (
+    MipEra,
+    cv_column,
+    dataset_model_for,
+    era_scope,
+    eras_in_scope,
+    mip_era_for,
+    source_id_datasets,
+)
 from ref_backend.models.model_runs import RunCounts
 
 
@@ -30,79 +40,76 @@ class ModelFacets(NamedTuple):
     dataset_count: int
 
 
-def _model_groups(mip_era: MipEra | None, source_id: str | None) -> CompoundSelect[Any] | Select[Any] | None:
-    """
-    Select every (execution group, model) pairing at the promoted version of each diagnostic.
+def _model_filter(mip_era: MipEra | None, **kwargs: Any) -> ExecutionGroupFilter:
+    return ExecutionGroupFilter(**era_scope(mip_era), **kwargs)
 
-    A group that ran several models appears once per model, so the counts answer
-    "what did this model take part in" rather than partitioning the groups.
 
-    Participation is read from the execution that decides the group, the same one the outcome
-    comes from. A retry run against different datasets would otherwise report the latest outcome
-    for a model that took no part in it.
-    A group whose deciding execution is retracted is left out, rather than falling back to an older run.
-    Only `source_id` is carried alongside, so a group naming one model across two eras or two
-    institutions still yields one row and is counted once.
-    """
-    latest = select_latest_executions().subquery()
+def _outcome(execution: Any) -> Any:
+    return case(
+        (execution.successful.is_(True), "successful"),
+        (execution.successful.is_(False), "failed"),
+        else_="running",
+    )
+
+
+def _model_datasets(mip_era: MipEra | None, source_id: str | None) -> Subquery:
+    """Select the id and `source_id` of every model dataset in scope, across the eras' dataset tables."""
     statements = []
     for era in eras_in_scope(mip_era):
         dataset_model = dataset_model_for(era)
         model_source_id = cv_column(dataset_model, "source_id")
-        statement = (
-            select(
-                models.ExecutionGroup.id.label("group_id"),
-                models.ExecutionGroup.diagnostic_id.label("diagnostic_id"),
-                model_source_id.label("source_id"),
-                models.Execution.id.label("execution_id"),
-            )
-            .join(models.Diagnostic, models.ExecutionGroup.diagnostic_id == models.Diagnostic.id)
-            .join(latest, latest.c.execution_group_id == models.ExecutionGroup.id)
-            .join(models.Execution, models.Execution.id == latest.c.execution_id)
-            .join(models.Execution.datasets.of_type(dataset_model))
-            .where(
-                models.ExecutionGroup.diagnostic_version == models.Diagnostic.promoted_version,
-                models.Execution.retracted.is_(False),
-                model_source_id.isnot(None),
-            )
-            .distinct()
+        statement = select(dataset_model.id.label("dataset_id"), model_source_id.label("source_id")).where(
+            model_source_id.isnot(None)
         )
         if source_id:
             statement = statement.where(model_source_id == source_id)
         statements.append(statement)
+    return (union(*statements) if len(statements) > 1 else statements[0]).subquery("model_datasets")
 
-    return union(*statements) if statements else None
 
+def _model_groups(mip_era: MipEra | None, source_id: str | None) -> Select[Any]:
+    """
+    Select every (execution group, model) pairing the group filter keeps, with the group's outcome.
 
-def _outcome(execution: Any) -> Any:
-    return case(
-        (execution.successful.is_(True), "successful"),
-        (execution.successful.is_(False), "failed"),
-        else_="running",
+    A group that ran several models appears once per model, so the counts answer
+    "what did this model take part in" rather than partitioning the groups.
+
+    Participation is read from the execution that decides the group, the same one the outcome
+    comes from. A retry run against different datasets would otherwise report the latest outcome
+    for a model that took no part in it.
+    A group whose deciding execution is retracted is left out, rather than falling back to an older run.
+    Only `source_id` is carried alongside, so a group naming one model across two eras or two
+    institutions still yields one row and is counted once.
+    """
+    # Joined once, because a select per era would look up every group's latest execution again.
+    datasets = _model_datasets(mip_era, source_id)
+    return (
+        select_execution_group_ids(_model_filter(mip_era))
+        .with_only_columns(
+            models.ExecutionGroup.id.label("group_id"),
+            models.ExecutionGroup.diagnostic_id.label("diagnostic_id"),
+            datasets.c.source_id,
+            models.Execution.id.label("execution_id"),
+            _outcome(models.Execution).label("outcome"),
+        )
+        .join(execution_datasets, execution_datasets.c.execution_id == models.Execution.id)
+        .join(datasets, datasets.c.dataset_id == execution_datasets.c.dataset_id)
+        .distinct()
     )
 
 
 def model_run_rows(
     session: Session, *, mip_era: MipEra | None = None, source_id: str | None = None
 ) -> list[ModelRunRow]:
     """Tally execution groups per model, diagnostic and outcome."""
-    groups = _model_groups(mip_era, source_id)
-    if groups is None:
-        return []
-
-    subquery = groups.subquery("model_groups")
-    execution = aliased(models.Execution)
-    outcome = _outcome(execution)
-
+    subquery = _model_groups(mip_era, source_id).subquery("model_groups")
     rows = session.execute(
         select(
             subquery.c.source_id,
             subquery.c.diagnostic_id,
-            outcome.label("outcome"),
+            subquery.c.outcome,
             func.count(distinct(subquery.c.group_id)).label("group_count"),
-        )
-        .join(execution, execution.id == subquery.c.execution_id)
-        .group_by(subquery.c.source_id, subquery.c.diagnostic_id, outcome)
+        ).group_by(subquery.c.source_id, subquery.c.diagnostic_id, subquery.c.outcome)
     ).all()
 
     return [ModelRunRow(*row) for row in rows]
@@ -167,22 +174,14 @@ def failed_runs(
 
     Returns the group, the id of the execution that decided the outcome, and the outcome itself.
     """
-    groups = _model_groups(mip_era, source_id)
-    if groups is None:
-        return []
-
-    subquery = groups.subquery("model_groups")
-    execution = aliased(models.Execution)
-    outcome = _outcome(execution)
-
-    rows = session.execute(
-        select(models.ExecutionGroup, execution.id, outcome.label("outcome"))
-        .select_from(subquery)
-        .join(models.ExecutionGroup, models.ExecutionGroup.id == subquery.c.group_id)
-        .join(execution, execution.id == subquery.c.execution_id)
-        .where(outcome != "successful")
-        .distinct()
-        .order_by(models.ExecutionGroup.updated_at.desc())
-    ).all()
+    filters = _model_filter(mip_era, successful=False, input_datasets=source_id_datasets(source_id))
+    statement: Select[Any] = (
+        select_execution_groups(filters).order_by(None).order_by(models.ExecutionGroup.updated_at.desc())
+    )
+    rows = session.execute(statement).tuples().all()
 
-    return [(row[0], row[1], row[2]) for row in rows]
+    return [
+        (group, execution.id, "failed" if execution.successful is False else "running")
+        for group, execution in rows
+        if execution
+    ]

backend/tests/perf/test_endpoints.py

@@ -91,9 +91,34 @@ class Endpoint:
     Endpoint(
         id="models.list",
         path="models/",
-        allowed_scans=frozenset({"execution", "cmip6_dataset", "cmip7_dataset"}),
+        allowed_scans=frozenset({"execution", "execution_group", "cmip6_dataset", "cmip7_dataset"}),
+    ),
+    Endpoint(
+        id="models.get",
+        path="models/{source_id}",
+        allowed_scans=frozenset({"execution", "execution_group", "obs4mips_dataset"}),
+    ),
+    Endpoint(
+        id="models.ensemble", path="models/{source_id}/ensemble", allowed_scans=frozenset({"execution"})
+    ),
+    Endpoint(
+        id="models.list.cmip6",
+        path="models/",
+        params={"mip_era": "CMIP6"},
+        allowed_scans=frozenset({"execution_group", "cmip6_dataset"}),
+    ),
+    Endpoint(
+        id="models.get.cmip6",
+        path="models/{source_id}",
+        params={"mip_era": "CMIP6"},
+        allowed_scans=frozenset({"execution_group"}),
+    ),
+    Endpoint(
+        id="models.ensemble.cmip6",
+        path="models/{source_id}/ensemble",
+        params={"mip_era": "CMIP6"},
+        allowed_scans=frozenset({"execution"}),
     ),
-    Endpoint(id="models.get", path="models/{source_id}", allowed_scans=frozenset({"execution"})),
     Endpoint(id="datasets.list", path="datasets/"),
     Endpoint(id="datasets.get", path="datasets/{dataset_slug}"),
 ]

backend/tests/test_core/test_model_runs.py

@@ -118,6 +118,7 @@ def test_model_groups_carries_only_the_keys_it_counts_on():
         "diagnostic_id",
         "source_id",
         "execution_id",
+        "outcome",
     ]