Merge pull request #862 from DataIntegrationGroup/fix/duplicate-rows-in-batch

fix(ingestion): stop duplicate instants reaching one INSERT

Author: jirhiker

automated_ingestion/ocotillo/loader.py

@@ -119,6 +119,14 @@ def load_observations(
     table = TransducerObservation.__table__
 
     for batch in _batched(records, batch_size):
+        result.rows_seen += len(batch)
+
+        # One row per instant within a statement. Postgres refuses an
+        # ON CONFLICT DO UPDATE that would touch the same row twice in one
+        # command, and a source can repeat a reading -- overlapping fetch
+        # windows, or a vendor logging the same instant twice. Keeping the last
+        # occurrence matches the upsert's own rule: a later value wins.
+        deduplicated = {record.observation_datetime: record for record in batch}
         rows = [
             {
                 "deployment_id": deployment_id,
@@ -128,9 +136,10 @@ def load_observations(
                 "release_status": release_status,
                 "data_maturity": data_maturity,
             }
-            for record in batch
+            for record in deduplicated.values()
         ]
-        result.rows_seen += len(rows)
+        if not rows:
+            continue
 
         statement = insert(table).values(rows)
         # DO UPDATE rather than DO NOTHING: a vendor may correct a reading, and

automated_ingestion/shared/windows.py

@@ -74,10 +74,17 @@ def iter_windows(start: int, end: int, span: int = DEFAULT_SPAN) -> Iterator[Win
         raise ValueError(f"Window span must be positive, got {span}.")
     if end < start:
         raise ValueError(f"End {end} precedes start {start}.")
+    # Windows must not share a boundary. Diver-HUB's ranges are inclusive at
+    # both ends -- "from start time up to and including end time" -- so
+    # [0, span] and [span, 2*span] both return the reading logged exactly at
+    # `span`. That duplicate reaches the loader in one batch and Postgres
+    # rejects the statement: "ON CONFLICT DO UPDATE command cannot affect row a
+    # second time".
     cursor = start
     while cursor < end:
-        yield Window(cursor, min(cursor + span, end))
-        cursor += span
+        chunk_end = min(cursor + span, end)
+        yield Window(cursor, chunk_end)
+        cursor = chunk_end + 1
 
 
 # ============= EOF =============================================

automated_ingestion/tests/test_windows.py

@@ -25,20 +25,32 @@
 )
 
 
-def test_windows_cover_the_range_without_gaps_or_overlap():
+def test_windows_do_not_share_a_boundary():
+    # Diver-HUB ranges are inclusive at both ends, so touching windows both
+    # return the reading logged exactly on the boundary. That duplicate reaches
+    # the loader in one batch and Postgres rejects the statement: "ON CONFLICT
+    # DO UPDATE command cannot affect row a second time".
     windows = list(iter_windows(0, 10 * DAY, span=3 * DAY))
     assert windows[0].start == 0
     assert windows[-1].end == 10 * DAY
     for earlier, later in zip(windows, windows[1:]):
-        assert earlier.end == later.start
+        assert later.start == earlier.end + 1
+
+
+def test_windows_leave_no_second_uncovered():
+    # The gap is exactly one second and timestamps are second-resolution, so
+    # nothing can fall between two windows.
+    windows = list(iter_windows(0, 10 * DAY, span=3 * DAY))
+    for earlier, later in zip(windows, windows[1:]):
+        assert later.start - earlier.end == 1
 
 
 def test_final_window_is_truncated_not_overshot():
     # Overshooting would ask the API for a future range, which is at best waste
     # and at worst a 400.
     windows = list(iter_windows(0, 10 * DAY, span=3 * DAY))
     assert windows[-1].end == 10 * DAY
-    assert windows[-1].span == DAY
+    assert all(w.end <= 10 * DAY for w in windows)
 
 
 def test_range_shorter_than_span_is_a_single_window():

tests/test_transducer_loader.py

@@ -268,4 +268,36 @@ def test_rows_with_no_recorded_maturity_still_update(loader_target):
         assert value == 99.0
 
 
+def test_duplicate_instants_in_one_batch_do_not_break_the_statement(loader_target):
+    """Reproduces the first live run's failure.
+
+    Postgres rejects an ON CONFLICT DO UPDATE that would touch the same row
+    twice in one command:
+
+        ON CONFLICT DO UPDATE command cannot affect row a second time
+
+    The existing idempotency test loads the same window in two separate
+    statements, which Postgres allows, so it could not catch this. A source can
+    repeat an instant -- overlapping fetch windows did, and a vendor may log one
+    twice.
+    """
+    deployment_id, parameter_id = loader_target
+    duplicated = _records(3) + _records(3, value=99.0)
+
+    with session_ctx() as session:
+        result = load_observations(
+            session, duplicated, deployment_id, parameter_id, "draft"
+        )
+        assert _count(session, deployment_id) == 3
+        assert result.rows_seen == 6
+
+        # The later value wins, matching the upsert's own rule.
+        values = session.scalars(
+            select(TransducerObservation.value)
+            .where(TransducerObservation.deployment_id == deployment_id)
+            .order_by(TransducerObservation.observation_datetime)
+        ).all()
+        assert values[0] == 99.0
+
+
 # ============= EOF =============================================