feat(ingestion): derive the watermark from Postgres - #848
Merged
Conversation
Aqueduct keeps watermarks in a GCS object because FROST cannot be queried cheaply for a maximum. Ocotillo's destination can, so the watermark is MAX(observation_datetime) for the series. That is a deliberate divergence. 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. A derived one cannot disagree with reality. It also turns "backfill never advances the normal watermark" from a rule to enforce into a property that cannot be violated -- re-loading a window behind the maximum cannot move a maximum forward. Tested in both directions anyway, older data and the same window twice, since the claim is load-bearing for task 4. Keyed by thing rather than 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 floor applies only to a series that has never been loaded, in either direction -- a floor ahead of the watermark does not win either, so it cannot be used as a backfill lever by accident. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
|
Your pull request is automatically being deployed to Dagster Cloud.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Task 3.5.
automated_ingestion/shared/watermark.py, seven tests.Stacked on #847 — the tests load observations through the upsert loader, so
this targets that branch rather than
staging. Merge #847 first and thisretargets cleanly.
Derived, not stored
Aqueduct keeps watermarks in a GCS object beside the raw zone, because FROST
cannot be queried cheaply for a maximum. Ocotillo's destination can, so the
watermark is simply
MAX(observation_datetime)for the series.That is a deliberate divergence, recorded in the module docstring so it reads as
a decision rather than an oversight. 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.
The payoff: "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. Task 4 leans on this, so it is
asserted anyway, in both directions: older data, and the same window twice.
Keyed by thing, not deployment
The plan said
(thing_id, parameter_id)and that is right, for a reason worthstating. 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.This is the same column confusion that made the 3.4 constraint wrong in the
plan, landing the other way round.
The floor is not a lever
resolve_startfalls back toinitial_start_dateonly for a series neverloaded. A test asserts a floor ahead of the watermark does not win either — so
it cannot be used to skip forward any more than to backfill. Re-fetching history
is what the backfill jobs are for.
Session sharing
PostgresWatermarkStoretakes the session the loader is using, so the watermarkreflects that session's committed state rather than another connection's
snapshot. That matters once a run loads and then asks where it got to.
🤖 Generated with Claude Code