Idempotency and Safe Retries

A pipeline step will run twice. Not might — will: a task that times out after committing, a worker that dies before reporting success, a retried DAG, an overlapping backfill. The question is only whether the second run does damage.

Setup

import duckdb
con = duckdb.connect("bookshop.duckdb")

con.execute("""
    create or replace table incoming as
    select * from (values
        (1001, 1, date '2026-01-04', 'completed', 25.50),
        (1002, 2, date '2026-01-04', 'completed', 12.00),
        (1003, 1, date '2026-01-04', 'returned',  40.00)
    ) t(order_id, customer_id, ordered_at, status, amount)
""")

The default is wrong

con.execute("create or replace table orders as select * from incoming where false")

def load_append():
    con.execute("insert into orders select * from incoming")

load_append()
load_append()

print(con.execute("select count(*) as rows, count(distinct order_id) as ids from orders").df().to_string(index=False))

 rows  ids
    6    3

Six rows, three orders. Nothing failed and nothing warned; every downstream sum is now double. This is the most common data bug there is, and it is created by the most obvious way to write a load.

Pattern 1: full overwrite

def load_overwrite():
    con.execute("create or replace table orders as select * from incoming")

load_overwrite()
load_overwrite()
print(con.execute("select count(*) as rows from orders").df().to_string(index=False))

 rows
    3

The simplest correct answer, and the right one whenever the source is small enough to reload in full — reference data, dimension tables, anything under a few million rows. It stops being viable when a full reload costs hours, and it cannot express “yesterday is final, today is still changing”.

Pattern 2: delete-insert by partition

con.execute("create or replace table orders as select * from incoming where false")

def load_partition(run_date):
    con.execute("begin transaction")
    con.execute("delete from orders where ordered_at = ?", [run_date])
    con.execute("insert into orders select * from incoming where ordered_at = ?", [run_date])
    con.execute("commit")

load_partition("2026-01-04")
load_partition("2026-01-04")
load_partition("2026-01-04")

print(con.execute("select ordered_at, count(*) from orders group by 1").df().to_string(index=False))

 ordered_at  count_star()
 2026-01-04             3

Three runs, three rows. The step deletes the slice it is about to write, so re-running replaces rather than adds — and it only touches one day, so a backfill of March does not disturb April.

Two things make it correct. The transaction means a crash between the delete and the insert rolls back rather than leaving an empty partition. And the delete predicate must match the insert exactly:

def load_broken(run_date):
    con.execute("delete from orders where ordered_at = ?", [run_date])
    con.execute("insert into orders select * from incoming")   # ← all dates, not one

load_broken("2026-01-04")
load_broken("2026-01-04")
print(con.execute("select count(*) as rows, count(distinct order_id) as ids from orders").df().to_string(index=False))

 rows  ids
    6    3

Duplicates again. The delete removed one day and the insert wrote everything, so any row outside that day accumulates. Whenever a delete-insert step duplicates, this mismatch is why.

Pattern 3: merge on a natural key

Delete-insert assumes rows belong to the partition being written. Late-arriving data breaks that — an order from Monday that turns up on Wednesday:

con.execute("""
    create or replace table incoming as
    select * from (values
        (1003, 1, date '2026-01-04', 'refunded',  40.00),   -- status changed
        (1007, 3, date '2026-01-04', 'completed', 18.40)    -- late arrival
    ) t(order_id, customer_id, ordered_at, status, amount)
""")

def load_merge():
    con.execute("""
        merge into orders t
        using incoming s on t.order_id = s.order_id
        when matched then update set status = s.status, amount = s.amount
        when not matched then insert values
            (s.order_id, s.customer_id, s.ordered_at, s.status, s.amount)
    """)

load_merge()
load_merge()

print(con.execute("select order_id, status, amount from orders order by order_id").df().to_string(index=False))

 order_id    status  amount
     1001 completed   25.50
     1002 completed   12.00
     1003  refunded   40.00
     1007 completed   18.40

Run it any number of times and the result is the same: 1003 updated, 1007 inserted once. MERGE is the general answer — it is what dbt’s incremental models, Delta Lake, Snowflake and BigQuery all use underneath.

The requirement is a genuine natural key. If order_id is not unique in the source, the merge either fails or picks a row arbitrarily:

con.execute("insert into incoming values (1007, 3, date '2026-01-04', 'returned', 18.40)")
load_merge()

duckdb.duckdb.InvalidInputException: MERGE with multiple source rows matching
the same target row is not allowed

A clear error rather than a coin flip. Deduplicate the source first:

con.execute("""
    create or replace table incoming_deduped as
    select * exclude (rn) from (
        select *, row_number() over (partition by order_id order by ordered_at desc) as rn
        from incoming
    ) where rn = 1
""")

Pattern 4: deduplicate on read

Sometimes you cannot control the write — an append-only event log, or a Kafka sink that guarantees at-least-once. Then the table holds duplicates by design and the view removes them:

con.execute("""
    create or replace table order_events as
    select * from (values
        (1001, 'created',   timestamp '2026-01-04 09:14:02', 1),
        (1001, 'created',   timestamp '2026-01-04 09:14:02', 1),   -- producer retry
        (1001, 'completed', timestamp '2026-01-04 09:18:41', 2),
        (1002, 'created',   timestamp '2026-01-04 09:20:00', 1)
    ) t(order_id, event, occurred_at, seq)
""")

con.execute("""
    create or replace view current_orders as
    select order_id, event, occurred_at
    from (
        select *, row_number() over (
            partition by order_id order by seq desc, occurred_at desc
        ) as rn
        from order_events
    )
    where rn = 1
""")

print(con.execute("select * from current_orders order by order_id").df().to_string(index=False))

 order_id     event         occurred_at
     1001 completed 2026-01-04 09:18:41
     1002   created 2026-01-04 09:20:00

The duplicate created event is still on disk and invisible to every reader. The ordering column matters — seq is monotonic per order, so a duplicate delivered late cannot overwrite a newer state. Ordering by wall-clock arrival time instead would.

Choosing

Side effects are the hard part

Table writes are the easy half. A step that also sends an email, calls an API or drops a file is not idempotent just because its table write is:

con.execute("""
    create table if not exists sent_notifications (
        idempotency_key varchar primary key,
        order_id bigint,
        sent_at timestamp
    )
""")

def notify(order_id, run_date):
    key = f"order-shipped:{order_id}:{run_date}"
    already = con.execute(
        "select count(*) from sent_notifications where idempotency_key = ?", [key]
    ).fetchone()[0]
    if already:
        print(f"skip {key}")
        return
    print(f"POST /notifications {key}")
    con.execute("insert into sent_notifications values (?, ?, now())", [key, order_id])

notify(1001, "2026-01-04")
notify(1001, "2026-01-04")
notify(1001, "2026-01-05")

POST /notifications order-shipped:1001:2026-01-04
skip order-shipped:1001:2026-01-04
POST /notifications order-shipped:1001:2026-01-05

The key is derived from the data and the run, not generated randomly — so the second attempt computes the same key and recognises itself. Most payment and messaging APIs accept an Idempotency-Key header for exactly this; when one does not, the local ledger above is the fallback.

A checklist

Before shipping any pipeline step, answer these:

1. What happens if this runs twice? Run it twice and check the row count.
2. What happens if it dies halfway? Is the write in one transaction?
3. What is the natural key? If there is not one, you cannot merge or deduplicate.
4. What happens on a backfill? Does re-running March touch April?
5. What side effects fire? Are they keyed on something stable?

Practice

 rows  ids
    6    3

 rows  ids
    6    3

duckdb.duckdb.InvalidInputException: MERGE with multiple source rows matching
the same target row is not allowed

POST /notifications order-shipped:1001:2026-01-04
skip order-shipped:1001:2026-01-04

Next: batch pipelines — scheduling, windows, and the boundary between runs.