From 5a50381b243569e64348ab1b5f95db3509ba8e64 Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 09:47:27 -0700 Subject: [PATCH] fix(ingestion): do not overwrite approved observations by default Ocotillo already holds 542,161 approved observations for 14 of the 38 San Acacia wells, from the AMPAPI transfer and running to August 2022. The upsert would have replaced every one of them with a vendor reading and downgraded it to provisional, silently, on any backfill covering that window. DO UPDATE exists so a vendor correction can revise our own provisional readings. Applying it to reviewed history from another source is a different act, and it should be one somebody chooses: overwrite_approved defaults to False and the conflict clause skips approved rows. Rows with NULL maturity still update. Unknown is not approved, and treating it as such would freeze the 394,086 legacy rows with no QC record against every future correction. Still owed: those AMPAPI rows were loaded under whatever datum that pipeline used, and ours are ground-surface centimetres converted to feet. Overlapping timestamps should be compared before any window covering 2016-2022 is loaded. Co-Authored-By: Claude Opus 5 --- automated_ingestion/ocotillo/loader.py | 32 ++++++++-- docs/automated-ingestion-pipeline-plan.md | 14 ++++ tests/test_transducer_loader.py | 78 +++++++++++++++++++++++ 3 files changed, 120 insertions(+), 4 deletions(-) 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 =============================================