fix(dags): build stg_gias_ukprn monthly, never daily (review)

Rebuilding the view in the daily build dropped the Ofsted views on it
(CASCADE), which dim_school reads: every daily run would have failed. A
selector test now rejects any build that drops views it then reads.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

pipeline/dags/school_data_pipeline.py

@@ -111,7 +111,7 @@ print(f'Validation passed: {{count}} GIAS rows')
     # where that DAG hasn't run (pipeline/tests/test_dag_selectors.py).
     dbt_build = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ gias_code_names+ stg_gias_ukprn --exclude int_ks2_with_lineage+ int_ks4_with_lineage+ stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_gias_establishments+ stg_gias_links+ gias_code_names+ --exclude int_ks2_with_lineage+ int_ks4_with_lineage+ stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense = BashOperator(
@@ -146,7 +146,7 @@ with DAG(
 
     dbt_build_ofsted = BashOperator(
         task_id="dbt_build",
-        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ stg_ofsted_fe_providers+ stg_ofsted_fe_inspections+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+ --exclude stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
+        bash_command=f"cd {PIPELINE_DIR}/transform && {DBT_BIN} build --profiles-dir . --target production --select stg_ofsted_inspections+ stg_ofsted_fe_providers+ stg_ofsted_fe_inspections+ stg_gias_ukprn+ int_ofsted_latest+ fact_ofsted_inspection+ dim_school+ --exclude stg_ees_ks4_destinations+ stg_ees_ks5_destinations+",
     )
 
     sync_typesense_ofsted = BashOperator(

pipeline/tests/test_dag_selectors.py

@@ -96,3 +96,41 @@ def test_selected_models_only_read_models_that_exist(dag_id):
     missing = {model: sorted(PARENTS[model] - available) for model in sorted(selected)
                if PARENTS[model] - available}
     assert missing == {}, f'{dag_id} builds models whose parents it never builds: {missing}'
+
+
+TABLE = re.compile(r"materialized\s*=\s*['\"]table['\"]")
+
+
+def views():
+    """Staging and intermediate models are views unless configured as tables."""
+    out = set()
+    for path in MODELS.rglob('*.sql'):
+        layer = path.relative_to(MODELS).parts[0]
+        if layer in ('staging', 'intermediate') and not TABLE.search(path.read_text()):
+            out.add(path.stem)
+    return out
+
+
+VIEWS = views()
+
+
+@pytest.mark.parametrize('dag_id', sorted(BUILDS))
+def test_rebuilt_relations_do_not_drop_views_the_build_reads(dag_id):
+    """dbt rebuilds a relation by dropping the old one with CASCADE, which drops
+    every view built on it. A view dropped that way and not rebuilt by the same
+    build is gone when that build reads it: the daily GIAS build rebuilt
+    stg_gias_ukprn and dropped the Ofsted views dim_school reads (audit H4)."""
+    selected = selected_models(BUILDS[dag_id], PARENTS)
+    children = defaultdict(set)
+    for model, ps in PARENTS.items():
+        for p in ps:
+            children[p].add(model)
+    dropped, stack = set(), list(selected)
+    while stack:
+        for child in children[stack.pop()]:
+            if child in VIEWS and child not in dropped:
+                dropped.add(child)
+                stack.append(child)
+    dropped -= selected
+    broken = {m: sorted(PARENTS[m] & dropped) for m in sorted(selected) if PARENTS[m] & dropped}
+    assert broken == {}, f'{dag_id} drops views it then reads: {broken}'

pipeline/transform/models/staging/stg_gias_ukprn.sql

@@ -3,6 +3,9 @@
 -- A model on the raw source rather than a column of stg_gias_establishments:
 -- a ref to that model would put the Ofsted models into the daily build's
 -- stg_gias_establishments+ selection (pipeline/tests/test_dag_selectors.py).
+-- Built by the monthly Ofsted DAG with the views on it, never by the daily
+-- one: rebuilding a view drops the views on it (CASCADE), and dim_school, which
+-- the daily build makes, reads int_ofsted_latest and int_ofsted_listed.
 
 select
     cast("URN" as integer)          as urn,