Idempotent pipelines: why re-running a job should be safe

Every pipeline fails sometimes: a network blip, an expired credential, a full disk, a server restarted mid-run. What happens next is decided by a question asked in advance: if this job runs again, what state is the data left in?

If the answer is "exactly the state a clean run would have produced", the job is idempotent, and recovery is just running it again. If the answer is "some rows twice" or "a mix of old and new", every failure becomes an incident.

The classic trap: append-only loads

The simplest load appends every batch to a table. It is also the one that is unsafe to repeat.

The script below loads the same two daily figures twice, once with an append and once with an upsert.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE daily_revenue (day TEXT PRIMARY KEY, revenue REAL NOT NULL)")


def load(rows):
    with conn:
        conn.executemany(
            """
            INSERT INTO daily_revenue (day, revenue) VALUES (?, ?)
            ON CONFLICT (day) DO UPDATE SET revenue = excluded.revenue
            """,
            rows,
        )


batch = [("2025-03-01", 120.0), ("2025-03-02", 95.5)]
load(batch)
load(batch)  # the job runs again after a retry
print(conn.execute("SELECT COUNT(*), SUM(revenue) FROM daily_revenue").fetchone())
load([("2025-03-02", 101.0)])  # a corrected figure arrives
print(conn.execute("SELECT * FROM daily_revenue ORDER BY day").fetchall())

# the version that is NOT safe to run twice
conn.execute("CREATE TABLE log_revenue (day TEXT, revenue REAL)")
for _ in range(2):
    conn.executemany("INSERT INTO log_revenue VALUES (?, ?)", batch)
print(conn.execute("SELECT COUNT(*), SUM(revenue) FROM log_revenue").fetchone())

(2, 215.5)
[('2025-03-01', 120.0), ('2025-03-02', 101.0)]
(4, 431.0)

The upsert table holds two rows and 215.5 after two runs, and the corrected figure simply replaced the old one. The append-only table holds four rows and 431.0, double the truth, and nothing in it reveals the mistake.

Three ways to make a load idempotent

1. Upsert on a natural key. As above: a key that identifies the fact (the day, or order number plus line) and INSERT ... ON CONFLICT DO UPDATE (or MERGE in warehouses that use that name). Best for row-level facts with a stable identity.

2. Replace a partition. Delete everything for the slice you are about to load and insert the new version, in one transaction. For a daily batch, "delete where day = X, then insert the rows for X". It also handles rows that disappeared from the source, which an upsert would leave behind. Best for whole-period batches.

3. Append, then deduplicate on read. Keep everything and let a view pick the latest row per key. It is simple to write and keeps history, at the cost of every reader needing the view.

Whichever you choose, the write must be atomic, so a crash leaves either the old state or the new state and never a half-written one.

What else makes a run non-repeatable

• Using "now" inside the job. WHERE created_at > now() - interval '1 day' gives a different answer each time it runs. Pass the date range in as a parameter, so a re-run (or a backfill) uses the same window.
• Generated ids and timestamps. A random UUID created during the run makes every attempt a new row. Derive ids from the data, for example a hash of the natural key.
• Side effects that cannot be undone. Sending an e-mail or calling a payment API is not repeatable. Record that it was done, and check the record first.
• Reading a source that changes. If the source can be edited between attempts, snapshot it or read up to a fixed watermark.
• Order-dependent logic. If the result depends on the order rows arrive, sort first or make the logic order-independent.

A test worth keeping

For each important pipeline, run it twice against a copy of the data and compare the output tables. They must be identical. Then run it for a past day and compare with what is already there. If both match, you can retry at 3 a.m. without waking anyone.