feat(ingestion): add the transducer unique constraint and upsert loader - #847
Merged
Conversation
The table had only an index on (deployment_id, parameter_id, observation_datetime), so nothing prevented inserting the same reading twice. That absence is what forces a delete-then-repost load: with no constraint to conflict on, a re-run can only avoid duplicates by removing what is there first, leaving a window where the data is missing and, if the run fails midway, leaving it missing for good. The constraint is on deployment_id, not thing_id as the plan said -- that column is on the block table, not this one. The existing index already used deployment_id, and the scope is right anyway: a deployment is a thing/sensor pairing, so two sensors on one well may legitimately report the same instant. The migration drops the old index, since the constraint creates an equivalent one and keeping both means maintaining two on every insert into the largest table in the schema. Verified up and down against a database with 88,666 rows. find_duplicate_observations.sql runs first: the migration fails on a table that already violates the constraint. It separates redundant copies from groups whose values disagree, because those are conflicting measurements rather than duplicates and collapsing them would discard a reading somebody recorded. The loader upserts with DO UPDATE rather than DO NOTHING, so a vendor correction is applied instead of silently ignored, and uses Core rather than ORM objects per AGENTS.md. The idempotency test runs against a real Postgres, because the claim depends on the database enforcing the constraint. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Coverage✅ 79.40% total — gate is 75%. Coverage for the Python files changed in this PR
|
Contributor
|
Your pull request is automatically being deployed to Dagster Cloud.
|
This was referenced Aug 19, 2026
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.4. Migration
a1b2c3d4e5f6, loader inautomated_ingestion/ocotillo/loader.py, and three tests against a realPostgres.
Why a constraint at all
transducer_observationhad only an index on(deployment_id, parameter_id, observation_datetime), so nothing prevented inserting the same reading twice.That absence is what forces a delete-then-repost load: with no constraint to
conflict on, a re-run can only avoid duplicates by removing what is already
there — which leaves a window where the data is missing, and if the run fails
midway, leaves it missing for good.
The plan named a column that does not exist
It called for
UniqueConstraint(thing_id, parameter_id, observation_datetime).thing_idis onTransducerObservationBlock, not on the observation table,which carries
deployment_id. The existing index already useddeployment_id.The scope is right on its own terms too: a deployment is a thing/sensor pairing,
so two sensors on one well may legitimately report the same instant.
Before merging — run the duplicate report
psql "..." -f automated_ingestion/sql/find_duplicate_observations.sqlThe migration fails on a table that already violates the constraint, and
finding that out halfway through a production migration is worse than not
starting.
The report separates redundant copies from groups whose
valuedisagrees. Thoseare not duplicates but conflicting measurements, and collapsing them silently
would discard a reading somebody recorded — so it reports rather than deletes.
The local development database was clean (0 duplicate groups in 88,666 rows).
That is encouraging and is not evidence about production.
Details worth review
The migration drops the old index. The unique constraint creates an
equivalent one, and keeping both means two indexes maintained on every insert
into the largest table in the schema. Verified up and down against the 88,666-row
database.
DO UPDATE, notDO NOTHING. A vendor correction arriving as a no-op wouldleave the old value in place while the run reported success — the worst of both
outcomes. There is a test.
Core, not ORM objects.
AGENTS.mdis explicit for high-volume tables:instantiating a mapped class per observation is what turns a backfill into an
hour-long run.
Batches commit individually (5,000 rows). An interrupted load keeps what it
wrote, and resuming simply rewrites those rows because the write is an upsert.
Testing
The idempotency claim depends on Postgres enforcing the constraint and on
ON CONFLICTresolving against it, so it cannot be shown with a stub. All threetests run against the test database. 49 ingestion tests still pass.
🤖 Generated with Claude Code