Idempotency, Atomicity, and Pipeline Restartability

-- BAD: plain INSERT — NOT idempotent
INSERT INTO silver.orders (order_id, status, amount)
VALUES (9284751, 'delivered', 380.00);
-- run this twice → two rows with order_id = 9284751

-- GOOD: upsert — idempotent
INSERT INTO silver.orders (order_id, status, amount, updated_at)
VALUES (9284751, 'delivered', 380.00, '2026-03-17 20:14:32')
ON CONFLICT (order_id) DO UPDATE SET
    status = EXCLUDED.status, amount = EXCLUDED.amount, updated_at = EXCLUDED.updated_at
WHERE silver.orders.updated_at < EXCLUDED.updated_at;
-- the WHERE clause stops a replayed OLDER record from overwriting a newer one
-- REQUIRES a UNIQUE constraint or PK on order_id — verify it exists:
SELECT constraint_name FROM information_schema.table_constraints
WHERE table_name = 'orders' AND constraint_type IN ('PRIMARY KEY', 'UNIQUE');

>>> run pipeline for 2026-03-17, twice in a row
SELECT COUNT(*) FROM silver.orders WHERE order_date = '2026-03-17';
-- 48,234  (identical after both runs — upsert did its job)

-- BAD: relative window — NOT idempotent
SELECT * FROM orders WHERE updated_at > NOW() - INTERVAL '15 minutes';
-- a run at 06:00 extracts from 05:45; a rerun at 06:10 extracts from 05:55 —
-- rows between 05:45 and 05:55 are silently missed on the rerun

-- GOOD: fixed window, upper bound stored at run start
SELECT * FROM orders
WHERE updated_at > '2026-03-17 05:45:00'   -- from checkpoint
  AND updated_at <= '2026-03-17 06:00:00'; -- fixed at run start, not re-computed on retry

# BAD: append — NOT idempotent (rerun adds duplicate rows to the same file)
# with open('s3://bucket/orders/2026-03-17.csv', 'a') as f: f.write(new_rows)

# GOOD: overwrite the partition — idempotent
df.write.mode('overwrite').partitionBy('order_date').parquet('s3://freshcart-lake/silver/orders')
# rerunning for 2026-03-17 overwrites the date=2026-03-17 partition —
# output is identical no matter how many times it runs

import hashlib

def create_payment_idempotency_key(payment_id: str, amount: float, ts: str) -> str:
    """Same inputs → same key every time → API recognises and ignores the duplicate."""
    payload = f'{payment_id}:{amount}:{ts}'
    return hashlib.sha256(payload.encode()).hexdigest()[:32]

key = create_payment_idempotency_key('pay_xxx', 380.00, '2026-03-17T20:14:32Z')
response = requests.post('https://api.stripe.com/v1/payments',
    headers={'X-Idempotency-Key': key, 'Authorization': f'Bearer {api_key}'},
    json={'amount': 38000, 'currency': 'USD'})
# a retry with the same key returns the SAME response — the payment is not duplicated

# Distributed dedup — Redis SET NX (atomic, safe for concurrent consumers)
def is_duplicate(event_id: str, redis_client) -> bool:
    result = redis_client.set(f'processed:{event_id}', '1', nx=True, ex=86400)
    return result is None   # None = key already existed = duplicate

# Database-level, for pipelines that must guarantee exactly-once:
CREATE TABLE IF NOT EXISTS pipeline.processed_events (
    event_id VARCHAR(100) PRIMARY KEY, processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

INSERT INTO pipeline.processed_events (event_id) VALUES ('evt_xxx')
ON CONFLICT (event_id) DO NOTHING RETURNING event_id;
-- returns a row  → first time seeing this event → process it
-- returns nothing → duplicate → skip it

>>> INSERT ... ON CONFLICT (event_id) DO NOTHING RETURNING event_id  (2nd delivery of evt_xxx)
(0 rows)
# empty result set — the consumer knows to skip processing entirely

# BAD: auto-commit per row — NOT atomic
conn.autocommit = True
for row in rows:
    cur.execute("INSERT INTO silver.orders ...", row)
# crash after row 23,000 of 50,000 → 23,000 rows in, 27,000 missing, no clean restart point

# GOOD: one transaction per batch — atomic
conn.autocommit = False
with conn:   # commits on exit, rolls back on exception
    for row in rows:
        cur.execute("INSERT INTO silver.orders ...", row)
    # crash mid-loop: the ENTIRE batch rolls back — destination unchanged, rerun is correct

# BETTER: bulk insert, 10-100× faster than a row loop
with conn:
    psycopg2.extras.execute_values(cur,
        "INSERT INTO silver.orders (order_id, status, amount) VALUES %s "
        "ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status",
        [(r['order_id'], r['status'], r['amount']) for r in rows], page_size=5000)

with conn:
    cur.execute("CREATE TABLE silver.store_master_new AS SELECT * FROM source.stores")
    cur.execute("ALTER TABLE silver.store_master RENAME TO store_master_old")
    cur.execute("ALTER TABLE silver.store_master_new RENAME TO store_master")
    # ↑ from this line, ALL queries see new data — zero window of empty/partial data
    cur.execute("DROP TABLE silver.store_master_old")
# COMMIT: rename becomes permanent

-- Snowflake equivalent (atomic DDL):
ALTER TABLE silver.store_master SWAP WITH silver.store_master_new;   -- instant, no downtime

Readers, at every instant during the swap:
before commit:  store_master_old (old data, via MVCC)
after commit:   store_master (new data)
NEVER visible:  an empty table, a partially-loaded table, or two tables at once

from pathlib import Path

def write_parquet_atomically(df, final_path: str) -> None:
    final, tmp = Path(final_path), Path(final_path).with_suffix('.tmp.parquet')
    try:
        df.to_parquet(tmp, compression='zstd', index=False)   # potentially slow
        tmp.rename(final)   # atomic on POSIX — readers see old OR new, never partial
    except Exception:
        if tmp.exists():
            tmp.unlink()
        raise

# S3: a single PUT is atomic (object exists fully or not at all).
# Use a distinct temp prefix for in-progress writes, then copy to final:
#   write to:  s3://bucket/tmp/run-{run_id}/part-001.parquet
#   copy to:   s3://bucket/bronze/orders/date=2026-03-17/part-001.parquet
#   delete:    s3://bucket/tmp/run-{run_id}/part-001.parquet
# downstream readers only scan the bronze/ prefix — never see in-progress tmp/ files

def write_with_validation(rows: list[dict], dest_conn, run_id: str) -> None:
    staging_table = f'silver.orders_staging_{run_id.replace("-", "_")}'
    try:
        # Phase 1: write to staging — can fail, production is unaffected
        with dest_conn:
            dest_conn.execute(f'CREATE TABLE {staging_table} AS SELECT * FROM silver.orders WHERE 1=0')
            psycopg2.extras.execute_values(dest_conn.cursor(),
                f'INSERT INTO {staging_table} VALUES %s', [tuple(r.values()) for r in rows])

        # Phase 2: validate staging BEFORE it ever touches production
        with dest_conn.cursor() as cur:
            cur.execute(f'SELECT COUNT(*) FROM {staging_table} WHERE order_amount < 0')
            if cur.fetchone()[0] > 0:
                raise ValueError('Staging has negative order amounts')

            cur.execute("SELECT AVG(daily_count) FROM (SELECT DATE(ingested_at) d, COUNT(*) daily_count "
                        "FROM silver.orders WHERE ingested_at > NOW() - INTERVAL '7 days' GROUP BY 1) c")
            avg_daily = cur.fetchone()[0] or 0
            if avg_daily > 0 and abs(len(rows) - avg_daily) / avg_daily > 0.5:
                raise ValueError(f'Staging row count {len(rows)} deviates >50% from 7-day average {avg_daily:.0f}')

        # Phase 3: validation passed — atomically promote staging to production
        with dest_conn:
            dest_conn.execute(f"""
                INSERT INTO silver.orders SELECT * FROM {staging_table}
                ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status,
                    order_amount = EXCLUDED.order_amount, updated_at = EXCLUDED.updated_at
                WHERE silver.orders.updated_at < EXCLUDED.updated_at
            """)
    except Exception:
        raise   # staging still exists for inspection, production is unchanged
    finally:
        try:
            dest_conn.execute(f'DROP TABLE IF EXISTS {staging_table}')
            dest_conn.commit()
        except Exception:
            pass   # best-effort cleanup

>>> write_with_validation(rows_with_one_negative_amount, conn, run_id)
ValueError: Staging has negative order amounts
# production silver.orders: untouched, still showing yesterday's correct data
# the staging table is dropped in the finally block regardless

COARSE (one checkpoint at end of run):
  Fails on row 9,847 of 10,000 → next run re-processes ALL 10,000 from scratch.
  Cost: O(run_size) lost. Complexity: low. Use for: fast runs (< 5 min).

MEDIUM (checkpoint after each batch):
  Fails on batch 8 of 10 → next run re-processes only batches 8-10 (3,000 rows).
  Cost: O(batch_size) lost. Complexity: medium. Use for: long runs (> 10 min).

  batch_watermark = since
  for batch in extract_batches(since, until):
      transform_and_load(batch)
      batch_watermark = batch[-1]['updated_at']
      save_watermark(batch_watermark)   # checkpoint after EACH batch

FreshCart's silver_orders pipeline: 10,000 rows, batch_size=1,000, fails on batch 8
Coarse:  next run re-extracts and re-processes 10,000 rows (~12 min)
Medium:  next run re-extracts and re-processes 3,000 rows (~4 min)
Both produce the identical final row count — medium is just faster to recover

def complete_order_UNSAFE(order_id: int, conn, kafka_producer, api_client):
    conn.execute("UPDATE silver.orders SET status='completed' WHERE order_id=%s", (order_id,))
    conn.commit()                                          # committed
    kafka_producer.produce('orders.completed', key=str(order_id), value={...})
    kafka_producer.flush()                                 # if this fails: DB done, Kafka not
    api_client.notify_delivery_service(order_id)           # if this fails: both above done
    # any retry now = duplicate Kafka message, or worse, a duplicate charge to the merchant

def complete_order_SAFE(order_id: int, run_id: str, conn, kafka_producer, api_client):
    conn.execute("""
        INSERT INTO silver.orders (order_id, status, completed_at) VALUES (%s, 'completed', NOW())
        ON CONFLICT (order_id) DO UPDATE SET status = 'completed', completed_at = EXCLUDED.completed_at
        WHERE silver.orders.status != 'completed'
    """, (order_id,))
    conn.commit()

    # enable.idempotence=True on the Kafka producer: retries never produce duplicates
    kafka_producer.produce('orders.completed', key=str(order_id),
        value={'order_id': order_id, 'idempotency_key': f'{run_id}:{order_id}'})

    idempotency_key = f'order-complete-{order_id}-{run_id[:8]}'
    api_client.notify_delivery_service(order_id=order_id, headers={'X-Idempotency-Key': idempotency_key})

CREATE TABLE pipeline.order_completion_sagas (
    order_id BIGINT PRIMARY KEY, run_id VARCHAR(36) NOT NULL,
    db_updated BOOLEAN NOT NULL DEFAULT FALSE, kafka_published BOOLEAN NOT NULL DEFAULT FALSE,
    api_notified BOOLEAN NOT NULL DEFAULT FALSE, completed_at TIMESTAMPTZ
);

def complete_order_with_saga(order_id: int, run_id: str, ...):
    saga = load_or_create_saga(order_id, run_id)
    if not saga.db_updated:
        update_db(order_id); mark_saga_step(order_id, 'db_updated')
    if not saga.kafka_published:
        publish_kafka(order_id); mark_saga_step(order_id, 'kafka_published')
    if not saga.api_notified:
        notify_api(order_id); mark_saga_step(order_id, 'api_notified')
    mark_saga_complete(order_id)

>>> complete_order_with_saga(9284751, run_id, ...)   # retried after step 2 failed
# db_updated=True already → skipped
# kafka_published=False → publishes now
# api_notified=False → notifies now
# no duplicate DB update, no duplicate charge — each step ran exactly once

def test_double_run_produces_same_row_count(self, test_db, test_dest):
    run_date = '2026-03-17'
    run_pipeline(run_date, source_conn=test_db, dest_conn=test_dest)
    count_after_run1 = test_dest.execute("SELECT COUNT(*) FROM silver.orders").fetchone()[0]

    run_pipeline(run_date, source_conn=test_db, dest_conn=test_dest)
    count_after_run2 = test_dest.execute("SELECT COUNT(*) FROM silver.orders").fetchone()[0]

    assert count_after_run1 == count_after_run2, (
        f'Row count changed on second run: {count_after_run1} → {count_after_run2}')

def test_rerun_after_source_update_uses_latest_values(self, test_db, test_dest):
    run_pipeline('2026-03-17', source_conn=test_db, dest_conn=test_dest)
    assert get_status(test_dest, 9284751) == 'placed'

    test_db.execute("UPDATE orders SET status='delivered', updated_at=NOW() WHERE order_id=9284751")
    reset_checkpoint_to_before_run1()

    run_pipeline('2026-03-17', source_conn=test_db, dest_conn=test_dest)
    assert get_status(test_dest, 9284751) == 'delivered'

def test_pipeline_recovers_correctly_after_mid_run_failure(self, test_db, test_dest):
    insert_test_orders(test_db, count=10_000)

    call_count = 0
    def upsert_that_fails_on_batch_4(rows, conn):
        nonlocal call_count
        call_count += 1
        if call_count == 4:
            raise RuntimeError('Simulated failure on batch 4')
        return original_upsert(rows, conn)

    with pytest.raises(RuntimeError):
        with patch('pipeline.load.upsert_batch', side_effect=upsert_that_fails_on_batch_4):
            run_pipeline('2026-03-17', source_conn=test_db, dest_conn=test_dest)

    count_after_failure = row_count(test_dest)
    assert 0 < count_after_failure < 10_000   # some batches landed, not all

    run_pipeline('2026-03-17', source_conn=test_db, dest_conn=test_dest)   # recovery run
    assert row_count(test_dest) == 10_000     # no duplicates, no gaps

def test_ten_runs_same_result(self, test_db, test_dest):
    results = []
    for i in range(10):
        reset_checkpoint_for_run('2026-03-17')
        run_pipeline('2026-03-17', source_conn=test_db, dest_conn=test_dest)
        count = row_count(test_dest)
        checksum = test_dest.execute("SELECT SUM(order_amount) FROM silver.orders").fetchone()[0]
        results.append((count, checksum))

    assert len(set(results)) == 1, (
        f'Pipeline is NOT idempotent — 10 runs produced {len(set(results))} different results')

$ pytest tests/test_idempotency.py -v
test_double_run_produces_same_row_count PASSED
test_rerun_after_source_update_uses_latest_values PASSED
test_pipeline_recovers_correctly_after_mid_run_failure PASSED
test_ten_runs_same_result PASSED
========================== 4 passed in 3.82s ===========================

At 07:15 AM, the finance team reports yesterday’s revenue figure shows $8,423,000 — exactly double the $4,211,500 expected from manual bank reconciliation. The data engineering team begins investigating.

-- Step 1: when did the doubling occur?
SELECT DATE(ingested_at), COUNT(*) row_count, SUM(order_amount) revenue
FROM silver.orders WHERE order_date = '2026-03-17' GROUP BY 1 ORDER BY 1;
-- 48,234 rows, $4,211,500 (morning load — correct)
-- 96,468 rows, $8,423,000 (evening — doubled!)

-- Step 2: duplicate order IDs?
SELECT order_id, COUNT(*) copies FROM silver.orders
WHERE order_date = '2026-03-17' GROUP BY order_id HAVING COUNT(*) > 1;
-- 48,234 rows returned — every single order_id has exactly 2 copies

-- Step 3: Airflow run history
SELECT dag_run_id, start_date, state FROM airflow.dag_run
WHERE dag_id = 'orders_pipeline_incremental' AND start_date::DATE = '2026-03-17';
-- shows TWO full-load runs at 18:00 and 18:15 — someone triggered a manual backfill

-- Step 4: the actual INSERT statement
SELECT query_text FROM snowflake.account_usage.query_history
WHERE query_text ILIKE '%INSERT INTO silver.orders%' AND start_time::DATE = '2026-03-17';
-- "INSERT INTO silver.orders SELECT * FROM orders_staging" — plain INSERT, no ON CONFLICT

-- IMMEDIATE: deduplicate
CREATE TABLE silver.orders_deduped AS
SELECT DISTINCT ON (order_id) * FROM silver.orders ORDER BY order_id, ingested_at DESC;
ALTER TABLE silver.orders RENAME TO orders_duplicated_backup;
ALTER TABLE silver.orders_deduped RENAME TO orders;

-- PERMANENT:
-- 1. INSERT → INSERT ... ON CONFLICT DO UPDATE
-- 2. ALTER TABLE silver.orders ADD CONSTRAINT uq_order_id UNIQUE (order_id);
-- 3. Add an idempotency test to CI (Part 07) that fails if a rerun changes row count
-- 4. max_active_runs=1, and require code review for manual backfills

SELECT COUNT(*), SUM(order_amount) FROM silver.orders WHERE order_date = '2026-03-17';
-- 48,234 rows, $4,211,500 ← correct

Total impact: 07:15 alert → 07:52 fully resolved (37 minutes).
Finance report delayed 52 minutes past SLA. Correct in production by 08:00 AM.

The incident happened because one failure mode — a manual trigger of the pipeline for an already-processed date — was never considered. The plain INSERT that worked fine for the first run created duplicates on the second. Adding ON CONFLICT DO UPDATE and a UNIQUE constraint took 15 minutes. The idempotency test would have caught this before the first production deployment.