diff --git a/pipeline/dags/school_data_pipeline.py b/pipeline/dags/school_data_pipeline.py index da4a4b6..b4b4196 100644 --- a/pipeline/dags/school_data_pipeline.py +++ b/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( diff --git a/pipeline/tests/test_dag_selectors.py b/pipeline/tests/test_dag_selectors.py index 0d15846..bd52187 100644 --- a/pipeline/tests/test_dag_selectors.py +++ b/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}' diff --git a/pipeline/transform/models/staging/stg_gias_ukprn.sql b/pipeline/transform/models/staging/stg_gias_ukprn.sql index ed351e3..c626694 100644 --- a/pipeline/transform/models/staging/stg_gias_ukprn.sql +++ b/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,