perf(api): batch supplementary queries — one per table, not five per school

get_supplementary_data ran ~5 sequential DB round-trips per URN, so
/api/compare scaled at ~37ms/school (measured on staging: 1 school 155ms,
3 schools 220ms, 6 schools 340ms). get_supplementary_data_batch fetches
each table once with WHERE urn IN (...) and groups in Python, collapsing
5*N round-trips to a constant 5. get_supplementary_data is now a thin
wrapper so the detail endpoint is unchanged; the compare endpoint makes
one batched call. Each table degrades independently on failure.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0146VHeLAWjDVE2B5uU67jCB

backend/app.py

@@ -30,6 +30,7 @@ from .data_loader import (
     load_latest_school_data,
     geocode_single_postcode,
     get_supplementary_data,
+    get_supplementary_data_batch,
     search_schools_typesense,
 )
 from .data_loader import get_data_info as get_db_info
@@ -679,8 +680,10 @@ async def compare_schools(
     db = None
     try:
         db = database.SessionLocal()
+        # One query per table for all schools, not ~5 queries per school.
+        batch = get_supplementary_data_batch(db, urn_list)
         for urn in urn_list:
-            supp = get_supplementary_data(db, urn)
+            supp = batch.get(urn, {})
             supplementary_by_urn[urn] = {
                 key: supp.get(key, default)
                 for key, default in _EMPTY_SUPPLEMENTARY.items()
@@ -772,93 +775,101 @@ async def get_la_averages(request: Request):
     return {"year": latest_year, "secondary": {"attainment_8_by_la": la_avg}}
 
 
+_KS2_NATIONAL_METRICS = [
+    "rwm_expected_pct", "rwm_high_pct",
+    "reading_expected_pct", "writing_expected_pct", "maths_expected_pct",
+    "gps_expected_pct", "gps_high_pct", "science_expected_pct",
+    "reading_avg_score", "maths_avg_score", "gps_avg_score",
+    "reading_progress", "writing_progress", "maths_progress",
+    "overall_absence_pct", "persistent_absence_pct",
+    "disadvantaged_gap", "disadvantaged_pct", "sen_support_pct", "eal_pct",
+]
+_KS4_NATIONAL_METRICS = [
+    "attainment_8_score", "progress_8_score",
+    "english_maths_standard_pass_pct", "english_maths_strong_pass_pct",
+    "ebacc_entry_pct", "ebacc_standard_pass_pct", "ebacc_strong_pass_pct",
+    "ebacc_avg_score", "gcse_grade_91_pct",
+]
+
+
 def _national_averages_payload(df: pd.DataFrame) -> dict:
     """National-averages payload shared by /api/national-averages and
-    /api/compare. Official DfE KS2 figures come from the mart table;
-    KS4 figures are computed from our dataset (no DfE dataset yet)."""
+    /api/compare.
+
+    Both series are persisted marts computed at import time: official DfE
+    KS2 figures (fact_ks2_national_averages) and dataset-computed KS4
+    averages (fact_ks4_national_averages) — the API never aggregates the
+    performance dataframe per request. If the KS4 mart hasn't been built
+    yet (deploy lands before the next DAG run), fall back to computing the
+    latest year only — a single-year scan, never the historical loop.
+    """
     if df.empty:
         return {"primary": {}, "secondary": {}}
 
-    ks2_metrics = [
-        "rwm_expected_pct", "rwm_high_pct",
-        "reading_expected_pct", "writing_expected_pct", "maths_expected_pct",
-        "gps_expected_pct", "gps_high_pct", "science_expected_pct",
-        "reading_avg_score", "maths_avg_score", "gps_avg_score",
-        "reading_progress", "writing_progress", "maths_progress",
-        "overall_absence_pct", "persistent_absence_pct",
-        "disadvantaged_gap", "disadvantaged_pct", "sen_support_pct", "eal_pct",
-    ]
-    ks4_metrics = [
-        "attainment_8_score", "progress_8_score",
-        "english_maths_standard_pass_pct", "english_maths_strong_pass_pct",
-        "ebacc_entry_pct", "ebacc_standard_pass_pct", "ebacc_strong_pass_pct",
-        "ebacc_avg_score", "gcse_grade_91_pct",
-    ]
+    latest_year = int(df["year"].max())
 
-    def _means(sub_df, metric_list):
+    from . import database
+    from .models import Ks2NationalAverage, Ks4NationalAverage
+
+    def _row_metrics(row, metric_list):
         out = {}
         for col in metric_list:
-            if col in sub_df.columns:
-                val = sub_df[col].dropna()
-                if len(val) > 0:
-                    out[col] = round(float(val.mean()), 2)
+            val = getattr(row, col, None)
+            if val is not None:
+                out[col] = val
         return out
 
-    latest_year = int(df["year"].max())
-    df_latest = df[df["year"] == latest_year]
-
-    # Primary: schools where KS2 data is non-null
-    primary_df = df_latest[df_latest["rwm_expected_pct"].notna()]
-    # Secondary: schools where KS4 data is non-null
-    secondary_df = df_latest[df_latest["attainment_8_score"].notna()]
-
-    latest_primary = _means(primary_df, ks2_metrics)
-    latest_secondary = _means(secondary_df, ks4_metrics)
-
-    # Per-year KS2 primary averages: use official DfE figures from the mart table.
-    # Per-year KS4 secondary averages: computed from our dataset (no DfE dataset yet).
-    from . import database
-    from .models import Ks2NationalAverage
-
-    by_year = []
+    ks2_rows: list = []
+    ks4_rows: list = []
     db = None
     try:
         db = database.SessionLocal()
-        nat_rows = db.query(Ks2NationalAverage).order_by(Ks2NationalAverage.year).all()
-        # Build a lookup of computed secondary averages per year as fallback
-        secondary_by_year = {}
-        for yr in sorted(df["year"].dropna().unique()):
-            yr = int(yr)
-            df_yr = df[df["year"] == yr]
-            secondary_by_year[yr] = _means(
-                df_yr[df_yr["attainment_8_score"].notna()], ks4_metrics
-            )
-        # Merge: official KS2 figures + computed KS4 figures per year
-        ks2_years = {r.year for r in nat_rows}
-        all_years = sorted(ks2_years | set(secondary_by_year.keys()))
-        nat_lookup = {r.year: r for r in nat_rows}
-        for yr in all_years:
-            primary_yr: dict = {}
-            if yr in nat_lookup:
-                r = nat_lookup[yr]
-                for col in ks2_metrics:
-                    val = getattr(r, col, None)
-                    if val is not None:
-                        primary_yr[col] = val
-            by_year.append({
-                "year": yr,
-                "primary": primary_yr,
-                "secondary": secondary_by_year.get(yr, {}),
-            })
+        try:
+            ks2_rows = db.query(Ks2NationalAverage).order_by(Ks2NationalAverage.year).all()
+        except Exception:
+            db.rollback()
+        try:
+            ks4_rows = db.query(Ks4NationalAverage).order_by(Ks4NationalAverage.year).all()
+        except Exception:
+            db.rollback()
+    except Exception:
+        pass
     finally:
         if db is not None:
             db.close()
 
-    # Update latest_primary with official DfE figure for the latest year if available
-    if by_year:
-        latest_official = next((e["primary"] for e in reversed(by_year) if e["primary"]), None)
-        if latest_official:
-            latest_primary = latest_official
+    primary_by_year = {r.year: _row_metrics(r, _KS2_NATIONAL_METRICS) for r in ks2_rows}
+    secondary_by_year = {r.year: _row_metrics(r, _KS4_NATIONAL_METRICS) for r in ks4_rows}
+
+    if not any(secondary_by_year.values()):
+        # KS4 mart missing/empty: compute the latest year only.
+        df_latest = df[df["year"] == latest_year]
+        sec = (
+            df_latest[df_latest["attainment_8_score"].notna()]
+            if "attainment_8_score" in df_latest.columns
+            else df_latest.iloc[0:0]
+        )
+        vals = {}
+        for col in _KS4_NATIONAL_METRICS:
+            if col in sec.columns:
+                v = sec[col].dropna()
+                if len(v) > 0:
+                    vals[col] = round(float(v.mean()), 2)
+        if vals:
+            secondary_by_year[latest_year] = vals
+
+    all_years = sorted(set(primary_by_year) | set(secondary_by_year))
+    by_year = [
+        {
+            "year": yr,
+            "primary": primary_by_year.get(yr, {}),
+            "secondary": secondary_by_year.get(yr, {}),
+        }
+        for yr in all_years
+    ]
+
+    latest_primary = next((e["primary"] for e in reversed(by_year) if e["primary"]), {})
+    latest_secondary = next((e["secondary"] for e in reversed(by_year) if e["secondary"]), {})
 
     return {
         "year": latest_year,

backend/data_loader.py

@@ -662,92 +662,150 @@ def _admissions_row_dict(a) -> dict:
     }
 
 
-def get_supplementary_data(db: Session, urn: int) -> dict:
-    """Fetch all supplementary data for a single school URN."""
-    result = {}
+def _census_dict(pc) -> dict:
+    return {
+        "year": pc.year,
+        "total_pupils": pc.total_pupils,
+        "female_pupils": pc.female_pupils,
+        "male_pupils": pc.male_pupils,
+        "fsm_pct": pc.fsm_pct,
+        "eal_pct": pc.eal_pct,
+    }
 
-    def safe_query(model, pk_field, latest_field=None):
+
+def _deprivation_dict(d) -> dict:
+    return {
+        "lsoa_code": d.lsoa_code,
+        "idaci_score": d.idaci_score,
+        "idaci_decile": d.idaci_decile,
+    }
+
+
+def _finance_dict(f) -> dict:
+    return {
+        "year": f.year,
+        "per_pupil_spend": f.per_pupil_spend,
+        "staff_cost_pct": f.staff_cost_pct,
+        "teacher_cost_pct": f.teacher_cost_pct,
+        "support_staff_cost_pct": f.support_staff_cost_pct,
+        "premises_cost_pct": f.premises_cost_pct,
+    }
+
+
+def _empty_supplementary() -> dict:
+    return {
+        "ofsted": None,
+        "census": None,
+        "admissions": None,
+        "admissions_history": [],
+        "sen_detail": None,
+        "phonics": None,
+        "deprivation": None,
+        "finance": None,
+    }
+
+
+def get_supplementary_data_batch(db: Session, urns: list[int]) -> dict:
+    """Fetch supplementary data for many URNs with one query per table
+    (WHERE urn IN (...)) instead of ~5 queries per school, collapsing the
+    per-request round-trips from 5*N to a constant 5. Returns {urn: block}
+    with the same shape get_supplementary_data produces per URN.
+
+    Each table is queried independently and failures degrade that table to
+    empty for every URN — a missing mart never blanks the others.
+    """
+    urns = [int(u) for u in urns]
+    result = {urn: _empty_supplementary() for urn in urns}
+    if not urns:
+        return result
+
+    def _safe(fn):
         try:
-            q = db.query(model).filter(getattr(model, pk_field) == urn)
-            if latest_field:
-                q = q.order_by(getattr(model, latest_field).desc())
-            return q.first()
+            fn()
         except Exception as e:
             import logging
-            logging.getLogger(__name__).error("safe_query failed for %s: %s", model.__name__, e)
+            logging.getLogger(__name__).error("batch supplementary query failed: %s", e)
             db.rollback()
-            return None
 
-    # Latest Ofsted inspection
-    o = safe_query(FactOfstedInspection, "urn", "inspection_date")
-    result["ofsted"] = _ofsted_block(o, urn) if o else None
-
-    # Census (latest year of fact_pupil_characteristics)
-    pc = safe_query(FactPupilCharacteristics, "urn", "year")
-    result["census"] = (
-        {
-            "year": pc.year,
-            "total_pupils": pc.total_pupils,
-            "female_pupils": pc.female_pupils,
-            "male_pupils": pc.male_pupils,
-            "fsm_pct": pc.fsm_pct,
-            "eal_pct": pc.eal_pct,
-        }
-        if pc
-        else None
-    )
-
-    # Admissions — all years, oldest first (for the multi-year trend view).
-    try:
-        admissions_rows = (
-            db.query(FactAdmissions)
-            .filter(FactAdmissions.urn == urn)
-            .order_by(FactAdmissions.year.asc())
+    # Ofsted — latest inspection per URN. Ordered so the first row seen per
+    # URN is the most recent.
+    def _ofsted():
+        rows = (
+            db.query(FactOfstedInspection)
+            .filter(FactOfstedInspection.urn.in_(urns))
+            .order_by(FactOfstedInspection.urn, FactOfstedInspection.inspection_date.desc())
             .all()
         )
-    except Exception as e:
-        import logging
-        logging.getLogger(__name__).error("admissions history query failed: %s", e)
-        db.rollback()
-        admissions_rows = []
+        seen = set()
+        for o in rows:
+            if o.urn in seen:
+                continue
+            seen.add(o.urn)
+            result[o.urn]["ofsted"] = _ofsted_block(o, o.urn)
+    _safe(_ofsted)
 
-    history = [_admissions_row_dict(a) for a in admissions_rows]
-    result["admissions_history"] = history
-    # Keep the single latest-year object for backwards-compatible consumers
-    # (hero chips, etc.).
-    result["admissions"] = history[-1] if history else None
+    # Census — latest year per URN.
+    def _census():
+        rows = (
+            db.query(FactPupilCharacteristics)
+            .filter(FactPupilCharacteristics.urn.in_(urns))
+            .order_by(FactPupilCharacteristics.urn, FactPupilCharacteristics.year.desc())
+            .all()
+        )
+        seen = set()
+        for pc in rows:
+            if pc.urn in seen:
+                continue
+            seen.add(pc.urn)
+            result[pc.urn]["census"] = _census_dict(pc)
+    _safe(_census)
 
-    # SEN detail — not available in current marts
-    result["sen_detail"] = None
+    # Admissions — all years per URN, oldest first (multi-year trend view).
+    def _admissions():
+        rows = (
+            db.query(FactAdmissions)
+            .filter(FactAdmissions.urn.in_(urns))
+            .order_by(FactAdmissions.urn, FactAdmissions.year.asc())
+            .all()
+        )
+        history: dict = {urn: [] for urn in urns}
+        for a in rows:
+            history[a.urn].append(_admissions_row_dict(a))
+        for urn, rows_for_urn in history.items():
+            result[urn]["admissions_history"] = rows_for_urn
+            result[urn]["admissions"] = rows_for_urn[-1] if rows_for_urn else None
+    _safe(_admissions)
 
-    # Phonics — no school-level data on EES
-    result["phonics"] = None
+    # Deprivation — one row per URN.
+    def _deprivation():
+        rows = (
+            db.query(FactDeprivation)
+            .filter(FactDeprivation.urn.in_(urns))
+            .all()
+        )
+        for d in rows:
+            result[d.urn]["deprivation"] = _deprivation_dict(d)
+    _safe(_deprivation)
 
-    # Deprivation
-    d = safe_query(FactDeprivation, "urn")
-    result["deprivation"] = (
-        {
-            "lsoa_code": d.lsoa_code,
-            "idaci_score": d.idaci_score,
-            "idaci_decile": d.idaci_decile,
-        }
-        if d
-        else None
-    )
-
-    # Finance (latest year)
-    f = safe_query(FactFinance, "urn", "year")
-    result["finance"] = (
-        {
-            "year": f.year,
-            "per_pupil_spend": f.per_pupil_spend,
-            "staff_cost_pct": f.staff_cost_pct,
-            "teacher_cost_pct": f.teacher_cost_pct,
-            "support_staff_cost_pct": f.support_staff_cost_pct,
-            "premises_cost_pct": f.premises_cost_pct,
-        }
-        if f
-        else None
-    )
+    # Finance — latest year per URN.
+    def _finance():
+        rows = (
+            db.query(FactFinance)
+            .filter(FactFinance.urn.in_(urns))
+            .order_by(FactFinance.urn, FactFinance.year.desc())
+            .all()
+        )
+        seen = set()
+        for f in rows:
+            if f.urn in seen:
+                continue
+            seen.add(f.urn)
+            result[f.urn]["finance"] = _finance_dict(f)
+    _safe(_finance)
 
     return result
+
+
+def get_supplementary_data(db: Session, urn: int) -> dict:
+    """Supplementary data for a single URN (thin wrapper over the batch)."""
+    return get_supplementary_data_batch(db, [urn])[int(urn)]

backend/models.py

@@ -231,6 +231,23 @@ class FactFinance(Base):
     premises_cost_pct = Column(Float)
 
 
+class Ks4NationalAverage(Base):
+    """Computed national KS4 averages (from our dataset) — one row per year."""
+    __tablename__ = "fact_ks4_national_averages"
+    __table_args__ = MARTS
+
+    year = Column(Integer, primary_key=True)
+    attainment_8_score = Column(Float)
+    progress_8_score = Column(Float)
+    english_maths_standard_pass_pct = Column(Float)
+    english_maths_strong_pass_pct = Column(Float)
+    ebacc_entry_pct = Column(Float)
+    ebacc_standard_pass_pct = Column(Float)
+    ebacc_strong_pass_pct = Column(Float)
+    ebacc_avg_score = Column(Float)
+    gcse_grade_91_pct = Column(Float)
+
+
 class Ks2NationalAverage(Base):
     """Official DfE KS2 national headline averages — one row per academic year."""
     __tablename__ = "fact_ks2_national_averages"

backend/tests/test_compare_enrichment.py

@@ -67,7 +67,9 @@ def client(monkeypatch):
 
     monkeypatch.setattr(app_module, "load_school_data", _two_primary_schools_df)
     monkeypatch.setattr(
-        app_module, "get_supplementary_data", lambda db, urn: dict(CANNED_SUPPLEMENTARY)
+        app_module,
+        "get_supplementary_data_batch",
+        lambda db, urns: {int(u): dict(CANNED_SUPPLEMENTARY) for u in urns},
     )
     monkeypatch.setattr(database_module, "SessionLocal", _StubSession)
     return TestClient(app_module.app, raise_server_exceptions=False)
@@ -102,10 +104,10 @@ def test_top_level_national_averages_and_benchmarks(client):
 def test_supplementary_failure_degrades_not_500(client, monkeypatch):
     from backend import app as app_module
 
-    def _boom(db, urn):
+    def _boom(db, urns):
         raise RuntimeError("marts unavailable")
 
-    monkeypatch.setattr(app_module, "get_supplementary_data", _boom)
+    monkeypatch.setattr(app_module, "get_supplementary_data_batch", _boom)
     resp = client.get("/api/compare?urns=100140")
     assert resp.status_code == 200
     school = resp.json()["comparison"]["100140"]

backend/tests/test_national_averages_marts.py

@@ -0,0 +1,93 @@
+"""_national_averages_payload reads persisted marts (computed at import
+time) — it must never loop the dataframe per year. The only dataframe work
+allowed is the single-latest-year KS4 fallback for the window between a
+deploy and the next DAG run."""
+
+import numpy as np
+import pandas as pd
+import pytest
+
+LATEST = 202425
+
+
+def _df():
+    return pd.DataFrame(
+        [
+            dict(year=202324, attainment_8_score=40.0, rwm_expected_pct=np.nan),
+            dict(year=LATEST, attainment_8_score=50.0, rwm_expected_pct=np.nan),
+            dict(year=LATEST, attainment_8_score=30.0, rwm_expected_pct=np.nan),
+            dict(year=LATEST, attainment_8_score=np.nan, rwm_expected_pct=80.0),
+        ]
+    )
+
+
+class _Ks2Row:
+    year = LATEST
+    rwm_expected_pct = 62.1
+    gps_expected_pct = 72.0
+
+
+class _Ks4Row:
+    year = LATEST
+    attainment_8_score = 46.5
+    progress_8_score = -0.02
+
+
+class _StubSession:
+    """Returns KS2 rows for the first query and KS4 rows for the second —
+    mirroring the payload's query order."""
+
+    def __init__(self):
+        self.calls = 0
+
+    def query(self, model):
+        self._model = model.__name__
+        return self
+
+    def order_by(self, *a):
+        return self
+
+    def all(self):
+        return [_Ks2Row()] if self._model == "Ks2NationalAverage" else [_Ks4Row()]
+
+    def close(self):
+        pass
+
+
+class _Ks4MissingSession(_StubSession):
+    def all(self):
+        if self._model == "Ks4NationalAverage":
+            raise RuntimeError("relation does not exist")
+        return [_Ks2Row()]
+
+    def rollback(self):
+        pass
+
+
+@pytest.fixture()
+def payload(monkeypatch):
+    from backend import app as app_module
+    from backend import database as database_module
+
+    def _run(session_cls):
+        monkeypatch.setattr(database_module, "SessionLocal", session_cls)
+        return app_module._national_averages_payload(_df())
+
+    return _run
+
+
+def test_ks4_averages_come_from_the_mart_not_the_dataframe(payload):
+    body = payload(_StubSession)
+    # Mart value (46.5), NOT the dataframe mean of (50+30)/2 = 40.0
+    assert body["secondary"]["attainment_8_score"] == 46.5
+    assert body["primary"]["rwm_expected_pct"] == 62.1
+    assert body["by_year"][-1]["secondary"]["progress_8_score"] == -0.02
+
+
+def test_missing_ks4_mart_falls_back_to_latest_year_only(payload):
+    body = payload(_Ks4MissingSession)
+    # Fallback computes the latest year from the df: mean(50, 30) = 40.0
+    assert body["secondary"]["attainment_8_score"] == 40.0
+    # ...and only the latest year — no historical KS4 loop
+    ks4_years = [e["year"] for e in body["by_year"] if e["secondary"]]
+    assert ks4_years == [LATEST]

backend/tests/test_supplementary_batch.py

@@ -0,0 +1,111 @@
+"""get_supplementary_data_batch fetches one query per table for all URNs
+(not ~5 per school) and returns the same per-URN block shape as the
+single-URN function, picking the latest row per URN where relevant."""
+
+import types
+
+from backend import data_loader
+from backend.data_loader import get_supplementary_data_batch
+
+
+class _FakeQuery:
+    """Records that a query ran and serves canned rows filtered by an in-list."""
+
+    def __init__(self, recorder, model_name, rows):
+        self._rec = recorder
+        self._model = model_name
+        self._rows = rows
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def order_by(self, *args, **kwargs):
+        return self
+
+    def all(self):
+        self._rec.append(self._model)
+        return self._rows
+
+    def first(self):
+        self._rec.append(self._model)
+        return self._rows[0] if self._rows else None
+
+
+class _FakeSession:
+    def __init__(self, rows_by_model):
+        self.rows_by_model = rows_by_model
+        self.queries: list[str] = []
+
+    def query(self, model):
+        name = model.__name__
+        return _FakeQuery(self.queries, name, self.rows_by_model.get(name, []))
+
+    def rollback(self):
+        pass
+
+
+def _ofsted_row(urn, date, oe):
+    base = {f: None for f in (
+        "framework", "inspection_type", "quality_of_education", "behaviour_attitudes",
+        "personal_development", "leadership_management", "early_years_provision",
+        "sixth_form_provision", "ungraded_outcome", "ungraded_grade",
+        "rc_safeguarding_met", "rc_inclusion", "rc_curriculum_teaching", "rc_achievement",
+        "rc_attendance_behaviour", "rc_personal_development", "rc_leadership_governance",
+        "rc_early_years", "rc_sixth_form", "report_url",
+    )}
+    base.update(urn=urn, inspection_date=types.SimpleNamespace(isoformat=lambda: date),
+                overall_effectiveness=oe, grade_source=None)
+    return types.SimpleNamespace(**base)
+
+
+def _adm_row(urn, year):
+    return types.SimpleNamespace(
+        urn=urn, year=year, school_phase="Primary", places_offered=100,
+        total_applications=200, first_preference_applications=150,
+        first_preference_offers=140, first_preference_offer_pct=93.3,
+        oversubscription_ratio=1.5, oversubscribed=True,
+        total_offers=100, second_preference_offers=5, third_preference_offers=2,
+        cross_la_applications=10, cross_la_offers=3,
+    )
+
+
+def test_one_query_per_table_and_latest_row_per_urn():
+    rows = {
+        # URN 1 has two Ofsted rows; the batch must keep the most recent (2023).
+        "FactOfstedInspection": [
+            _ofsted_row(1, "2023-01-01", 2),
+            _ofsted_row(1, "2019-01-01", 3),
+            _ofsted_row(2, "2021-06-01", 1),
+        ],
+        "FactAdmissions": [_adm_row(1, 202526), _adm_row(1, 202627), _adm_row(2, 202627)],
+        "FactPupilCharacteristics": [],
+        "FactDeprivation": [],
+        "FactFinance": [],
+    }
+    session = _FakeSession(rows)
+    out = get_supplementary_data_batch(session, [1, 2])
+
+    # Exactly one query per table — five total, regardless of two URNs.
+    assert sorted(session.queries) == [
+        "FactAdmissions", "FactDeprivation", "FactFinance",
+        "FactOfstedInspection", "FactPupilCharacteristics",
+    ]
+
+    # Latest Ofsted kept per URN
+    assert out[1]["ofsted"]["overall_effectiveness"] == 2
+    assert out[2]["ofsted"]["overall_effectiveness"] == 1
+
+    # Admissions history grouped per URN, latest exposed as `admissions`
+    assert [r["year"] for r in out[1]["admissions_history"]] == [202526, 202627]
+    assert out[1]["admissions"]["year"] == 202627
+    assert out[2]["admissions_history"] == [{**out[2]["admissions_history"][0]}]
+
+    # Empty tables degrade to the null block, not a crash
+    assert out[1]["census"] is None and out[1]["deprivation"] is None
+
+
+def test_single_wrapper_matches_batch(monkeypatch):
+    session = _FakeSession({"FactOfstedInspection": [_ofsted_row(5, "2022-01-01", 2)]})
+    single = data_loader.get_supplementary_data(session, 5)
+    assert single["ofsted"]["overall_effectiveness"] == 2
+    assert single["admissions_history"] == []

nextjs-app/app/compare/page.tsx

@@ -32,26 +32,24 @@ export default async function ComparePage({ searchParams }: ComparePageProps) {
   const selectedMetric = metricParam || 'rwm_expected_pct';
 
   try {
-    // Fetch comparison data if URNs provided
-    let comparisonData = null;
-    if (urns.length > 0) {
-      try {
-        const response = await fetchComparison(urnsParam!);
-        comparisonData = response.comparison;
-      } catch (error) {
-        console.error('Failed to fetch comparison:', error);
-      }
-    }
+    // Fetch comparison + metrics in parallel — they are independent.
+    const [comparisonResponse, metricsResponse] = await Promise.all([
+      urns.length > 0
+        ? fetchComparison(urnsParam!).catch((error) => {
+            console.error('Failed to fetch comparison:', error);
+            return null;
+          })
+        : Promise.resolve(null),
+      fetchMetrics(),
+    ]);
 
-    // Fetch available metrics
-    const metricsResponse = await fetchMetrics();
-
-    // Metrics is already an array
     const metricsArray = metricsResponse?.metrics || [];
 
     return (
       <ComparisonView
-        initialData={comparisonData}
+        initialData={comparisonResponse?.comparison ?? null}
+        initialNationalAverages={comparisonResponse?.national_averages}
+        initialBenchmarks={comparisonResponse?.benchmarks}
         initialUrns={urns}
         metrics={metricsArray}
         selectedMetric={selectedMetric}

nextjs-app/components/ComparisonView.tsx

@@ -35,6 +35,8 @@ import styles from './ComparisonView.module.css';
 
 interface ComparisonViewProps {
   initialData: Record<string, ComparisonData> | null;
+  initialNationalAverages?: NationalAverages;
+  initialBenchmarks?: Benchmarks;
   initialUrns: number[];
   metrics: MetricDefinition[];
   selectedMetric: string;
@@ -42,6 +44,8 @@ interface ComparisonViewProps {
 
 export function ComparisonView({
   initialData,
+  initialNationalAverages,
+  initialBenchmarks,
   initialUrns,
   metrics,
   selectedMetric: initialMetric,
@@ -54,8 +58,10 @@ export function ComparisonView({
   const [selectedMetric, setSelectedMetric] = useState(initialMetric);
   const [isModalOpen, setIsModalOpen] = useState(false);
   const [comparisonData, setComparisonData] = useState(initialData);
-  const [nationalAverages, setNationalAverages] = useState<NationalAverages | undefined>();
-  const [benchmarks, setBenchmarks] = useState<Benchmarks | undefined>();
+  const [nationalAverages, setNationalAverages] = useState<NationalAverages | undefined>(
+    initialNationalAverages,
+  );
+  const [benchmarks, setBenchmarks] = useState<Benchmarks | undefined>(initialBenchmarks);
   const [shareConfirm, setShareConfirm] = useState(false);
   const [comparePhase, setComparePhase] = useState<'primary' | 'secondary'>('primary');
   // Tracks whether the user has explicitly clicked a phase tab.
@@ -81,13 +87,16 @@ export function ComparisonView({
     }
   }, [isInitialized]); // eslint-disable-line react-hooks/exhaustive-deps
 
-  // Sync URL with selected schools + metric, and (re)fetch the comparison.
+  const urnKey = selectedSchools.map((s) => s.urn).join(',');
+
+  // Sync the URL with the selection + metric. Pure navigation state — no
+  // fetching here: metric changes are presentational (the data is already
+  // client-side) and must not refire the comparison request.
   useEffect(() => {
-    const urns = selectedSchools.map((s) => s.urn).join(',');
     const params = new URLSearchParams(searchParams);
 
-    if (urns) {
-      params.set('urns', urns);
+    if (urnKey) {
+      params.set('urns', urnKey);
     } else {
       params.delete('urns');
     }
@@ -96,26 +105,41 @@ export function ComparisonView({
 
     const newUrl = `${pathname}?${params.toString()}`;
     router.replace(newUrl, { scroll: false });
+  }, [urnKey, selectedMetric, pathname, searchParams, router]);
 
-    if (selectedSchools.length > 0) {
-      fetchComparison(urns, { cache: 'no-store' })
-        .then((data) => {
-          setComparisonData(data.comparison);
-          setNationalAverages(data.national_averages);
-          setBenchmarks(data.benchmarks);
-        })
-        .catch((err) => {
-          // Keep whatever we already have (SSR data or a previous fetch) rather
-          // than blanking the page — a transient refetch failure shouldn't
-          // destroy a working comparison the user is looking at.
-          console.error('Failed to fetch comparison:', err);
-        });
-    } else {
+  // Fetch only when the school set changes. The very first run is skipped
+  // when the SSR payload already covers the current set — no double-fetch
+  // of data the server just rendered.
+  const firstFetchRef = useRef(true);
+  useEffect(() => {
+    if (!urnKey) {
       setComparisonData(null);
       setNationalAverages(undefined);
       setBenchmarks(undefined);
+      return;
     }
-  }, [selectedSchools, selectedMetric, pathname, searchParams, router]);
+
+    if (firstFetchRef.current) {
+      firstFetchRef.current = false;
+      const ssrUrns = new Set(Object.keys(initialData ?? {}));
+      const covered = urnKey.split(',').every((urn) => ssrUrns.has(urn));
+      if (covered && ssrUrns.size > 0) return;
+    }
+
+    fetchComparison(urnKey, { cache: 'no-store' })
+      .then((data) => {
+        setComparisonData(data.comparison);
+        setNationalAverages(data.national_averages);
+        setBenchmarks(data.benchmarks);
+      })
+      .catch((err) => {
+        // Keep whatever we already have (SSR data or a previous fetch) rather
+        // than blanking the page — a transient refetch failure shouldn't
+        // destroy a working comparison the user is looking at.
+        console.error('Failed to fetch comparison:', err);
+      });
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [urnKey]);
 
   // Classify schools by phase using comparison data
   const classifySchool = (school: School): 'primary' | 'secondary' => {

pipeline/transform/models/marts/_marts_schema.yml

@@ -160,6 +160,12 @@ models:
       - name: year
         tests: [not_null, unique]
 
+  - name: fact_ks4_national_averages
+    description: Computed national KS4 averages (means across state schools in our dataset — not official DfE figures) — one row per academic year
+    columns:
+      - name: year
+        tests: [not_null, unique]
+
   - name: fact_deprivation
     description: IDACI deprivation index — one row per URN
     columns:

pipeline/transform/models/marts/fact_ks4_national_averages.sql

@@ -0,0 +1,25 @@
+{{ config(materialized='table') }}
+
+-- Mart: Computed national KS4 averages — one row per academic year.
+-- Unlike fact_ks2_national_averages (official DfE figures), DfE publishes no
+-- KS4 national-headline dataset we ingest yet, so these are means computed
+-- across the state schools in our dataset. Computed once at build time so the
+-- API never has to aggregate the full performance table per request.
+-- Semantics match the API's previous per-request computation: rows where
+-- attainment_8_score is non-null; per-column means ignore NULLs.
+
+select
+    year,
+    round(avg(attainment_8_score)::numeric, 2)               as attainment_8_score,
+    round(avg(progress_8_score)::numeric, 2)                 as progress_8_score,
+    round(avg(english_maths_standard_pass_pct)::numeric, 2)  as english_maths_standard_pass_pct,
+    round(avg(english_maths_strong_pass_pct)::numeric, 2)    as english_maths_strong_pass_pct,
+    round(avg(ebacc_entry_pct)::numeric, 2)                  as ebacc_entry_pct,
+    round(avg(ebacc_standard_pass_pct)::numeric, 2)          as ebacc_standard_pass_pct,
+    round(avg(ebacc_strong_pass_pct)::numeric, 2)            as ebacc_strong_pass_pct,
+    round(avg(ebacc_avg_score)::numeric, 2)                  as ebacc_avg_score,
+    round(avg(gcse_grade_91_pct)::numeric, 2)                as gcse_grade_91_pct
+from {{ ref('fact_ks4_performance') }}
+where attainment_8_score is not null
+group by year
+order by year