diff --git a/automated_ingestion/ocotillo/loader.py b/automated_ingestion/ocotillo/loader.py index d867f66c6..7b831bf82 100644 --- a/automated_ingestion/ocotillo/loader.py +++ b/automated_ingestion/ocotillo/loader.py @@ -89,12 +89,27 @@ def load_observations( release_status: str, batch_size: int = DEFAULT_BATCH_SIZE, data_maturity: str = DEFAULT_DATA_MATURITY, + overwrite_approved: bool = False, ) -> LoadResult: """Upsert observations, committing per batch. ``records`` are ``ObservationRecord`` values from an adapter; resolving a source's point identifier to a deployment belongs to reference-data bootstrapping, not here, so the caller supplies the ids. + + ``overwrite_approved`` guards data somebody has already reviewed. By default + a row whose ``data_maturity`` is ``approved`` is left alone: the upsert exists + so a vendor correction can revise *our* provisional readings, not so a + re-fetch can quietly replace Bureau-approved history with a vendor's numbers + and downgrade it to provisional. + + This is not hypothetical. Fourteen of the thirty-eight San Acacia wells + already hold 542,161 approved observations from the AMPAPI transfer, running + to August 2022. A Mode A backfill over that window would have overwritten + every one of them. + + Setting it to True is a deliberate act: it says the incoming data is better + than what was reviewed, which is a judgement a person should make. """ from sqlalchemy.dialects.postgresql import insert @@ -121,17 +136,26 @@ def load_observations( # DO UPDATE rather than DO NOTHING: a vendor may correct a reading, and # a correction arriving as a no-op would leave the old value in place # while the run reported success. - statement = statement.on_conflict_do_update( - index_elements=[ + conflict_kwargs: dict[str, Any] = { + "index_elements": [ "deployment_id", "parameter_id", "observation_datetime", ], - set_={ + "set_": { "value": statement.excluded.value, "data_maturity": statement.excluded.data_maturity, }, - ) + } + if not overwrite_approved: + # IS DISTINCT FROM rather than != so NULL maturity still updates: + # a row with no recorded status has not been reviewed, and treating + # unknown as approved would freeze 394,086 legacy rows against every + # future correction. + conflict_kwargs["where"] = table.c.data_maturity.is_distinct_from( + "approved" + ) + statement = statement.on_conflict_do_update(**conflict_kwargs) session.execute(statement) session.commit() diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 72aaff0e6..028321bd7 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -325,6 +325,20 @@ Built. Migration `a1b2c3d4e5f6`, loader in `automated_ingestion/ocotillo/loader. ⬜ Run the duplicate report against production and staging before applying the migration. The local development database was clean — 0 duplicate groups in 88,666 rows — which is encouraging and not evidence about production. +### Existing San Acacia data — measured 2026-08-19 + +Ocotillo already holds transducer observations for **14 of the 38 wells**: 542,161 rows from the AMPAPI transfer, running 2016-07-08 to **2022-08-03**. They carry a real QC status, so `data_maturity` backfilled them as `approved`. + +Consequences, all of which the earlier plan assumed away: + +- **The watermark starts at 2022-08-03 for those 14**, not the 2015 floor, so a normal run fetches a four-year gap rather than a decade. The other 24 wells do start at the floor. +- **A backfill would have overwritten them.** The upsert's `DO UPDATE` was written for vendor corrections to our own provisional readings; against approved AMPAPI history it would have replaced 542,161 reviewed values with vendor numbers *and* downgraded them to provisional. `load_observations` now refuses to touch an `approved` row unless `overwrite_approved=True` is passed deliberately. +- **A datum comparison is still owed.** Those rows came from AMPAPI under whatever convention that pipeline used; ours are Diver-HUB ground-surface centimetres converted to feet. Before any window overlapping 2016–2022 is loaded, a few coinciding timestamps should be compared. Same failure shape as the `WaterLevelReference` question: plausible numbers, wrong meaning. + +Rows with `NULL` maturity still update. Unknown is not approved, and treating it as such would freeze the 394,086 legacy rows that have no QC record against every future correction. + +**The wider table**, for context: 2,180,989 approved, 7,351 provisional, 394,086 NULL. The NULL cohort is 176 deployments on a single parameter spanning 2016 to February 2025 with no AMPAPI provenance at all — a separate network, and **none of the 38 San Acacia wells are in it**. Worth identifying independently of this work. + ### 3.5 — Watermark from Postgres Built. `automated_ingestion/shared/watermark.py`, seven tests. diff --git a/tests/test_transducer_loader.py b/tests/test_transducer_loader.py index ec3f85355..8913713da 100644 --- a/tests/test_transducer_loader.py +++ b/tests/test_transducer_loader.py @@ -190,4 +190,82 @@ def test_maturity_must_be_a_lexicon_term(loader_target): session.rollback() +def test_approved_rows_are_not_overwritten(loader_target): + # 14 of the 38 San Acacia wells already hold 542,161 approved observations + # from the AMPAPI transfer. A backfill over that window must not replace + # reviewed values with a vendor's numbers. + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations( + session, + _records(1, value=10.0), + deployment_id, + parameter_id, + "draft", + data_maturity="approved", + ) + load_observations( + session, _records(1, value=99.0), deployment_id, parameter_id, "draft" + ) + row = session.execute( + select( + TransducerObservation.value, TransducerObservation.data_maturity + ).where(TransducerObservation.deployment_id == deployment_id) + ).one() + assert (row.value, row.data_maturity) == (10.0, "approved") + + +def test_approved_rows_can_be_overwritten_deliberately(loader_target): + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations( + session, + _records(1, value=10.0), + deployment_id, + parameter_id, + "draft", + data_maturity="approved", + ) + load_observations( + session, + _records(1, value=99.0), + deployment_id, + parameter_id, + "draft", + overwrite_approved=True, + ) + value = session.scalar( + select(TransducerObservation.value).where( + TransducerObservation.deployment_id == deployment_id + ) + ) + assert value == 99.0 + + +def test_rows_with_no_recorded_maturity_still_update(loader_target): + # 394,086 legacy rows have NULL maturity. Unknown is not approved, and + # treating it as such would freeze them against every future correction. + deployment_id, parameter_id = loader_target + with session_ctx() as session: + load_observations( + session, _records(1, value=10.0), deployment_id, parameter_id, "draft" + ) + session.execute( + TransducerObservation.__table__.update() + .where(TransducerObservation.deployment_id == deployment_id) + .values(data_maturity=None) + ) + session.commit() + + load_observations( + session, _records(1, value=99.0), deployment_id, parameter_id, "draft" + ) + value = session.scalar( + select(TransducerObservation.value).where( + TransducerObservation.deployment_id == deployment_id + ) + ) + assert value == 99.0 + + # ============= EOF =============================================