Making a load job idempotent

Reruns happen anyway

A load job will run twice sooner or later. The scheduler retries after a timeout that turned out to be a slow network. An engineer clears a failed task after fixing an upstream bug. A backfill covers a week that overlaps yesterday's successful run. Each of these is ordinary. A job that only behaves when it runs exactly once will eventually double a day's revenue in someone's report.

Idempotent means that running the job again on the same input leaves the target in the same state as running it once. The useful unit to reason about is the target table. Most of the decisions below follow from one question: after this run, what does the table look like, and would a second run change it?

Delete then insert by partition

The simplest pattern is to own a partition. Each run deletes every row in the partition it is responsible for, then inserts freshly computed rows. A second run deletes what the first one wrote and writes the same thing again.

This works well when source data arrives in complete daily slices. It fails quietly when a day is still receiving events, because the first run may have seen only part of them. Either wait until the partition is closed before the job touches it, or accept that a later rerun is the only thing that fixes an early one.

The delete and the insert must share a transaction. Otherwise a crash between them leaves an empty partition that looks like a legitimate result. Where the warehouse cannot wrap many statements in one transaction, write to a staging table and swap it in with a single rename or partition exchange.

MERGE on a natural key

When rows do not map cleanly onto partitions, match on a key instead. The natural key is the set of columns that identify a real-world thing, such as the order ID assigned by the source system. Generated surrogate keys are a poor choice here, because every rerun produces new ones and the target fills with duplicates.

MERGE INTO orders AS target
USING staging_orders AS source
    ON target.order_id = source.order_id
WHEN MATCHED AND target.updated_at < source.updated_at THEN
    UPDATE SET status = source.status,
               amount = source.amount,
               updated_at = source.updated_at
WHEN NOT MATCHED THEN
    INSERT (order_id, status, amount, updated_at)
    VALUES (source.order_id, source.status, source.amount, source.updated_at);

The condition on updated_at matters. Without it, an older file that gets reprocessed late can overwrite a newer row. Most engines also reject a MERGE where several source rows match the same target row, so deduplicate staging first and keep the latest version of each key. That step is easy to skip when test fixtures are clean, and it is the step that breaks in production.

Deterministic run keys

Anything the job writes that depends on the clock will differ between attempts. A run identifier built from a random UUID makes every retry look like a new run to downstream systems, which may then count it twice. Derive identifiers from inputs instead: the run date, the source name, a checksum of the input file. Two attempts over the same input then produce the same keys.

Keep now() out of transforms

A transform that calls now() or CURRENT_TIMESTAMP gives each rerun a different answer for rows that have not changed. A loaded_at column is fine as metadata. Keep it out of filters, joins, and derived values. If the logic needs a reference time, pass it in as a parameter.

A run_date parameter in Python

Here the job takes its date from the caller, not from the clock:

from datetime import date, datetime, timedelta

def load_daily_orders(run_date: date, conn) -> int:
    """Rebuild one day of orders. Safe to run any number of times."""
    start = datetime.combine(run_date, datetime.min.time())
    end = start + timedelta(days=1)
    with conn:  # commits on success, rolls back on error
        with conn.cursor() as cur:
            cur.execute(
                "DELETE FROM daily_orders WHERE order_date = %s",
                (run_date,),
            )
            cur.execute(
                """
                INSERT INTO daily_orders (order_date, order_id, amount)
                SELECT %s, order_id, amount
                FROM raw_orders
                WHERE created_at >= %s AND created_at < %s
                """,
                (run_date, start, end),
            )
            return cur.rowcount

The scheduler passes run_date explicitly, and a backfill is just a loop over dates. The function does not know whether this is the first attempt or the fifth. The with conn: block is the transaction boundary, so a failure halfway through leaves the previous contents of that day in place rather than an empty day.

Testing by running twice

The test is easy to write and worth keeping. Load a small fixture that includes a duplicate order and one event with a late timestamp. Run the job for a fixed date, record the row count and a checksum of the partition, run it again, and assert that both values match. Then change one input row and run a third time. Only that row's output should change.

Passing this test does not prove the job is correct. Failing it proves there is a bug, and that bug is far easier to find on a quiet afternoon with fixture data than during a backfill.