diff --git a/automated_ingestion/shared/watermark.py b/automated_ingestion/shared/watermark.py new file mode 100644 index 000000000..3dfcdfebb --- /dev/null +++ b/automated_ingestion/shared/watermark.py @@ -0,0 +1,108 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Where a series got to, asked of the data rather than of a sidecar. + +Aqueduct keeps watermarks in a GCS object beside the raw zone, because its +destination is FROST and cannot be queried cheaply for a maximum. Ocotillo's +destination is Postgres, so the watermark is simply +``MAX(observation_datetime)`` for the series. + +**This is a deliberate divergence, not an oversight.** A stored watermark is a +second source of truth about what was loaded, and the two drift: a load that +half-succeeds, or a sidecar write that fails after the rows commit, leaves the +watermark claiming more or less than the data holds. Deriving it means the +answer cannot disagree with reality — and it makes a backfill safe by +construction, since re-loading an old window cannot move a maximum forward. + +**Keyed by thing, not by deployment.** Observations carry ``deployment_id``, but +a series outlives its hardware: replacing a diver creates a new deployment for +the same well, and a watermark keyed to the deployment would report nothing for +the new one and re-fetch the entire history. The query joins through +``deployment`` to ask the question the pipeline actually has -- how far along is +this well's depth-to-water record. +""" + +from datetime import datetime +from typing import Any, Protocol + + +class WatermarkStore(Protocol): + """Where a series has been loaded up to.""" + + def get(self, thing_id: int, parameter_id: int) -> datetime | None: + """Latest observation for the series, or ``None`` if never loaded.""" + ... + + +class PostgresWatermarkStore: + """Reads the watermark from the observations themselves. + + Takes the session the loader is using, so the watermark reflects that + session's committed state rather than a separate connection's snapshot. + """ + + def __init__(self, session: Any) -> None: + self._session = session + + def get(self, thing_id: int, parameter_id: int) -> datetime | None: + from sqlalchemy import func, select + + from db.deployment import Deployment + from db.transducer import TransducerObservation + + return self._session.scalar( + select(func.max(TransducerObservation.observation_datetime)) + .join( + Deployment, + Deployment.id == TransducerObservation.deployment_id, + ) + .where(Deployment.thing_id == thing_id) + .where(TransducerObservation.parameter_id == parameter_id) + ) + + +class InMemoryWatermarkStore: + """For tests, and for reasoning about a run without a database.""" + + def __init__(self, watermarks: dict[tuple[int, int], datetime] | None = None): + self._watermarks = dict(watermarks or {}) + + def get(self, thing_id: int, parameter_id: int) -> datetime | None: + return self._watermarks.get((thing_id, parameter_id)) + + def set(self, thing_id: int, parameter_id: int, value: datetime) -> None: + self._watermarks[(thing_id, parameter_id)] = value + + +def resolve_start( + store: WatermarkStore, + thing_id: int, + parameter_id: int, + floor: datetime, +) -> datetime: + """Where the next fetch should begin. + + ``floor`` applies only to a series that has never been loaded. It is not a + backfill lever: lowering it will not re-fetch history for a series whose + watermark has already advanced past it, because the watermark wins whenever + one exists. Re-fetching history is what the backfill jobs are for. + """ + watermark = store.get(thing_id, parameter_id) + return watermark if watermark is not None else floor + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index f7dd5225c..6213285e2 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -292,10 +292,18 @@ Built. Migration `a1b2c3d4e5f6`, loader in `automated_ingestion/ocotillo/loader. ### 3.5 — Watermark from Postgres -- Keep Aqueduct's `WatermarkStore` interface; Postgres implementation returns `MAX(observation_datetime)` for a `(thing_id, parameter_id)`, read in the same session as the write. No GCS sidecar for normal runs. -- Backfill never advances the normal watermark implicitly — inherent with upsert, but asserted in a test. -- In-memory implementation kept for tests. First-ever run for a series falls back to the `initial_start_date` floor. -- Divergence from Aqueduct recorded in the module docstring, so it reads as a decision not an oversight. +Built. `automated_ingestion/shared/watermark.py`, seven tests. + +- ✅ `PostgresWatermarkStore` returns `MAX(observation_datetime)` for the series, read through the loader's own session so it reflects that session's committed state rather than another connection's snapshot. +- ✅ `InMemoryWatermarkStore` for tests and for reasoning about a run without a database. +- ✅ `resolve_start` falls back to the `initial_start_date` floor only for a series that has never been loaded. A test asserts a floor *ahead* of the watermark does not win either — the floor is not a backfill lever in any direction. +- ✅ The divergence from Aqueduct is in the module docstring, so it reads as a decision rather than an oversight. + +**Keyed by thing, not deployment.** This section said `(thing_id, parameter_id)` and that turns out to be right for a reason worth stating: observations carry `deployment_id`, but a series outlives its hardware. Replacing a diver creates a new deployment for the same well, and a watermark keyed to the deployment would report nothing for the new one and re-fetch the entire history. The query joins through `deployment` to ask the question the pipeline actually has. + +**Why derive rather than store.** A stored watermark is a second source of truth about what was loaded, and the two drift — a half-succeeded load, or a sidecar write that fails after the rows commit, leaves it claiming more or less than the data holds. Aqueduct stores one because FROST cannot be queried cheaply for a maximum; Postgres can. + +The payoff is that "backfill never advances the normal watermark" stops being a rule to enforce and becomes a property that cannot be violated: re-loading a window behind the maximum cannot move a maximum forward. Asserted anyway, in two directions — older data, and the same window twice. --- diff --git a/tests/test_watermark.py b/tests/test_watermark.py new file mode 100644 index 000000000..e2606de1e --- /dev/null +++ b/tests/test_watermark.py @@ -0,0 +1,131 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Watermark behaviour, including the property that makes backfill safe. +""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import delete, select + +from automated_ingestion.ocotillo.loader import load_observations +from automated_ingestion.ocotillo.structs import ObservationRecord +from automated_ingestion.shared.watermark import ( + InMemoryWatermarkStore, + PostgresWatermarkStore, + resolve_start, +) +from db.engine import session_ctx +from db.parameter import Parameter +from db.transducer import TransducerObservation + +START = datetime(2026, 1, 1, tzinfo=timezone.utc) +FLOOR = datetime(2015, 1, 1, tzinfo=timezone.utc) + + +@pytest.fixture() +def series(sensor_to_water_well_thing_deployment): + """A deployment, its thing, and a parameter, cleaned up afterwards.""" + deployment = sensor_to_water_well_thing_deployment + with session_ctx() as session: + parameter_id = session.scalar(select(Parameter.id).limit(1)) + yield deployment.id, deployment.thing_id, parameter_id + session.execute( + delete(TransducerObservation).where( + TransducerObservation.deployment_id == deployment.id + ) + ) + session.commit() + + +def _records(count, start=START, value=10.0): + return [ + ObservationRecord( + external_point_id="sanacaciareach-40", + observation_datetime=start + timedelta(minutes=5 * i), + value=value, + units="ft", + ) + for i in range(count) + ] + + +class TestPostgresWatermark: + def test_unloaded_series_has_no_watermark(self, series): + deployment_id, thing_id, parameter_id = series + with session_ctx() as session: + store = PostgresWatermarkStore(session) + assert store.get(thing_id, parameter_id) is None + + def test_watermark_is_the_latest_observation(self, series): + deployment_id, thing_id, parameter_id = series + with session_ctx() as session: + load_observations( + session, _records(10), deployment_id, parameter_id, "draft" + ) + store = PostgresWatermarkStore(session) + assert store.get(thing_id, parameter_id) == START + timedelta(minutes=45) + + def test_backfilling_older_data_does_not_advance_it(self, series): + # The property that makes backfill safe: a watermark derived from the + # data cannot be moved forward by re-loading a window behind it. A + # stored watermark has to be defended against this; a derived one + # cannot have the problem. + deployment_id, thing_id, parameter_id = series + with session_ctx() as session: + load_observations( + session, _records(10), deployment_id, parameter_id, "draft" + ) + store = PostgresWatermarkStore(session) + before = store.get(thing_id, parameter_id) + + older = _records(10, start=START - timedelta(days=365)) + load_observations(session, older, deployment_id, parameter_id, "draft") + + assert store.get(thing_id, parameter_id) == before + + def test_reloading_the_same_window_does_not_move_it(self, series): + deployment_id, thing_id, parameter_id = series + with session_ctx() as session: + load_observations( + session, _records(5), deployment_id, parameter_id, "draft" + ) + store = PostgresWatermarkStore(session) + before = store.get(thing_id, parameter_id) + load_observations( + session, _records(5), deployment_id, parameter_id, "draft" + ) + assert store.get(thing_id, parameter_id) == before + + +class TestResolveStart: + def test_floor_applies_only_to_a_new_series(self): + assert resolve_start(InMemoryWatermarkStore(), 1, 2, FLOOR) == FLOOR + + def test_watermark_wins_over_the_floor(self): + # The floor is not a backfill lever: lowering it must not re-fetch + # history for a series that has already advanced past it. + store = InMemoryWatermarkStore({(1, 2): START}) + assert resolve_start(store, 1, 2, FLOOR) == START + + def test_a_floor_ahead_of_the_watermark_does_not_win_either(self): + store = InMemoryWatermarkStore({(1, 2): START}) + ahead = START + timedelta(days=365) + assert resolve_start(store, 1, 2, ahead) == START + + +# ============= EOF =============================================