Idempotent Upsert Patterns for Late-Arriving Telemetry

When an IoT gateway re-delivers a buffered batch or a device resends a corrected reading, the same telemetry row can hit your hypertable two, three, or a dozen times — and the write path has to converge on exactly one authoritative value without inflating row counts. The tool for that on TimescaleDB is INSERT ... ON CONFLICT, but with one non-negotiable constraint: because a hypertable is partitioned on its time column, any unique index that backs the conflict target must include that time column. This page turns re-delivery into a deterministic decision — new row, duplicate, or correction — and shows the SQL and psycopg v3 code that resolves each case exactly once.

# Input Profiling: What to Establish Before You Write

Idempotency is only definable relative to a natural key — the set of columns that identifies one logical measurement. For telemetry that key is almost always (device_id, time, metric): one device, at one instant, reporting one metric, yields one value. Profile these three things before choosing a strategy:

• The natural key and its uniqueness — confirm that (device_id, time, metric) genuinely identifies a single reading. If a device can emit two distinct values for the same metric at the same microsecond, the key is incomplete and no upsert can converge it.
• Delivery guarantees of the transport — MQTT QoS 1 and most gateway store-and-forward buffers are at-least-once: they will re-send on reconnect and after ACK loss. This is the dominant source of duplicates and the reason ON CONFLICT DO NOTHING alone is often enough.
• Correction frequency — how often a corrected value arrives for an already-stored key. Firmware that back-fills calibrated readings, or edge processors that revise a reading after a smoothing pass, produce genuine updates. If corrections are rare (<0.1% of rows) the update path can be lean; if common, it drives connection and lock planning.

The interaction with late arrival matters: a re-delivered row may target a time that falls in an older chunk, possibly one already compressed. That is where the hypertable-specific mechanics of the out-of-order data insertion path intersect with upserts, and it is covered under edge cases below.

# Comparison: Three Strategies Under the Time-Column Constraint

The defining rule on a hypertable is that a UNIQUE constraint or unique index must include every partitioning column — at minimum the time dimension. You cannot declare UNIQUE (device_id, metric) on a hypertable and expect it to enforce or serve as a conflict target; PostgreSQL rejects it because the index would have to span chunks the planner prunes by time. So the conflict target is always (device_id, time, metric) — time is not optional.

The unique index that backs every one of these must be created explicitly — TimescaleDB does not add it for you:

-- The conflict target MUST include the time partitioning column.
-- device_id + metric alone is REJECTED on a hypertable.
CREATE UNIQUE INDEX IF NOT EXISTS telemetry_natural_key
    ON telemetry (device_id, time, metric);

# DO NOTHING — drop re-deliveries, keep first write

INSERT INTO telemetry (device_id, time, metric, value)
VALUES (%s, %s, %s, %s)
ON CONFLICT (device_id, time, metric) DO NOTHING;

This is exactly-once for duplicates but ignores corrections — a resent-with-a-new-value row is silently discarded. Use it when the transport is at-least-once but values never change after first emission.

# DO UPDATE — let the latest value win

INSERT INTO telemetry (device_id, time, metric, value)
VALUES (%s, %s, %s, %s)
ON CONFLICT (device_id, time, metric)
DO UPDATE SET value = EXCLUDED.value
-- Guard the update so an identical re-delivery does NOT churn the heap:
WHERE telemetry.value IS DISTINCT FROM EXCLUDED.value;

The WHERE clause is the quiet win: without it, every duplicate performs a no-op heap update, generating dead tuples and VACUUM pressure. With it, true duplicates fall through as DO NOTHING and only genuine corrections write.

# Staging + MERGE — set-based reconciliation for big batches

-- 1. COPY the raw batch into an unlogged staging table (fast, crash-disposable).
CREATE UNLOGGED TABLE IF NOT EXISTS telemetry_stage (LIKE telemetry);
TRUNCATE telemetry_stage;
-- 2. Reconcile the whole batch in one set-based statement.
MERGE INTO telemetry AS t
USING (
    SELECT DISTINCT ON (device_id, time, metric)
           device_id, time, metric, value
    FROM telemetry_stage
    ORDER BY device_id, time, metric, ingested_at DESC  -- last write wins
) AS s
ON  t.device_id = s.device_id
AND t.time      = s.time
AND t.metric    = s.metric
WHEN MATCHED AND t.value IS DISTINCT FROM s.value
    THEN UPDATE SET value = s.value
WHEN NOT MATCHED
    THEN INSERT (device_id, time, metric, value)
         VALUES (s.device_id, s.time, s.metric, s.value);

The DISTINCT ON collapses intra-batch duplicates before the MERGE ever touches the hypertable — a batch that contains the same key three times becomes one row, so the target table is probed once, not three times.

# psycopg v3 batch upsert

For the row-at-a-batch pattern, executemany over an ON CONFLICT statement is idempotent and pipeline-friendly in psycopg v3. For heavy batches, the COPY-then-MERGE path moves an order of magnitude more rows per second:

import psycopg
from psycopg.rows import dict_row

UPSERT = """
    INSERT INTO telemetry (device_id, time, metric, value)
    VALUES (%(device_id)s, %(time)s, %(metric)s, %(value)s)
    ON CONFLICT (device_id, time, metric)
    DO UPDATE SET value = EXCLUDED.value
    WHERE telemetry.value IS DISTINCT FROM EXCLUDED.value
"""


def upsert_batch(conn_str: str, rows: list[dict]) -> None:
    """Idempotent batch upsert: re-runnable with the same input, same result."""
    with psycopg.connect(conn_str) as conn:
        with conn.cursor() as cur:
            cur.executemany(UPSERT, rows)  # pipelined in psycopg v3
        conn.commit()


def copy_then_merge(conn_str: str, rows: list[dict]) -> None:
    """High-volume path: COPY into staging, reconcile with one MERGE."""
    with psycopg.connect(conn_str, row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute("TRUNCATE telemetry_stage")
            with cur.copy(
                "COPY telemetry_stage (device_id, time, metric, value, ingested_at) "
                "FROM STDIN"
            ) as copy:
                for r in rows:
                    copy.write_row(
                        (r["device_id"], r["time"], r["metric"],
                         r["value"], r["ingested_at"])
                    )
            cur.execute(MERGE_SQL)  # the MERGE statement shown above
        conn.commit()

Both functions are safe to call again with the same input: the second run finds every key already present, updates nothing (values match), and inserts nothing. That re-runnable property is what makes them idempotent.

# Worked Example: A Late Batch of Nine Rows

A gateway reconnects after a five-minute outage and flushes a buffered batch. Suppose the hypertable already holds these two rows for device d-17:

The re-delivered batch carries nine rows. Three are exact duplicates of what is stored, one is a correction, and the rest are genuinely new:

Running the DO UPDATE upsert with the IS DISTINCT FROM guard:

• Rows 1, 2, 3 conflict on the key; values match, so the WHERE suppresses the write — zero heap churn.
• Row 4 conflicts on (d-17, 12:00:00, temp_c); value 21.9 differs from stored 21.4, so it updates in place. The 12:00:00 reading now reads 21.9 and there is still exactly one row for that key.
• Rows 5–9 find no conflict and insert.

The batch of nine collapses to five inserts and one update. Re-run the identical batch and it collapses to zero writes — the definition of convergence. Row count for d-17 went from 2 to 7, never to 9 or 11.

# Edge Cases & When to Deviate

• Upserting into a compressed chunk — a conflict target on a compressed chunk cannot be resolved in place; recent TimescaleDB versions transparently stage such writes, but an ON CONFLICT DO UPDATE against already-compressed data can error or force a decompression. For corrections that land in cold chunks, either decompress the target chunk first (decompress_chunk), route the correction through the staging-MERGE path so the reconcile runs after decompression, or defer it to a scheduled backfill. The reconnect-driven variant of this is handled in automating backfill for IoT gateway reconnects.
• Conflict-target limitations on hypertables — the conflict target must exactly match the columns of a unique index that includes time. You cannot use a partial unique index or an expression index that omits the partitioning column as a conflict arbiter. If your natural key is (device_id, metric) semantically, you still must add time to the index and accept that two readings at different instants are distinct rows.
• High-volume cost versus plain insert — every ON CONFLICT row pays for an index probe that a bare INSERT skips. On append-only hot paths where the transport is genuinely exactly-once, plain inserts are faster; reserve upserts for streams that actually re-deliver. When batches are large and mixed, the COPY-then-MERGE path amortizes the probe cost across a set-based statement and wins decisively over per-row upserts.
• Time-column drift — if a device resends with a slightly different timestamp (clock skew, re-quantized bucket), it will not conflict and you get a near-duplicate. Normalize timestamps to a fixed resolution before upserting, or the natural key will not match across re-deliveries.

# Verification

Two checks prove the invariant. First, that no key appears twice — a clean idempotency guarantee. Second, that a known correction actually landed:

-- 1. No duplicate natural keys anywhere in the hypertable.
SELECT device_id, time, metric, COUNT(*) AS n
FROM telemetry
GROUP BY device_id, time, metric
HAVING COUNT(*) > 1;
-- Expect zero rows. A non-empty result means the unique index is
-- missing on some chunk or the conflict target is wrong.

-- 2. Confirm the correction was applied, not appended.
SELECT device_id, time, metric, value
FROM telemetry
WHERE device_id = 'd-17'
  AND time = '2026-07-17 12:00:00+00'
  AND metric = 'temp_c';
-- Expect a single row with value = 21.9 (the corrected value).

Confirm the enforcing index exists on the hypertable and covers the time column via the catalog:

-- The unique index that backs the conflict target, per chunk-parent table.
SELECT indexrelid::regclass AS index_name,
       indisunique,
       pg_get_indexdef(indexrelid) AS definition
FROM pg_index
WHERE indrelid = 'telemetry'::regclass
  AND indisunique;
-- The definition must list the time column; if it does not, upserts
-- on a hypertable will fail to enforce uniqueness across chunks.

If the duplicate query returns rows, the most common cause is that the unique index was created before the table became a hypertable and did not include time, or a bulk load bypassed ON CONFLICT entirely. Rebuild the index with the time column and re-run the reconcile. For streams where corrections are the norm rather than the exception, weigh whether a periodic full reconcile via incremental versus full refresh strategies on any dependent continuous aggregate is cheaper than fine-grained upserts on every write.

# Related

• Automating Backfill for IoT Gateway Reconnects — replaying buffered batches after an outage without creating duplicates
• Handling Out-of-Order Data Insertion in TimescaleDB — how late rows route to older and compressed chunks
• Incremental vs Full Refresh Strategies — keeping aggregates consistent after corrections land
• Chunk Compression Scheduling & Automation — why upserts into cold chunks need a decompression path
• Data Retention, Compression & Lifecycle Automation — the parent guide covering the full ingest-to-retention lifecycle

← Back to Python ETL Orchestration