Idempotent Merge/Upsert Patterns, Batch Identity, Deduplication, and Retry Safety

Learning outcomes

After a worker crash, AtlasMart's change batch is delivered again. Two events are duplicated even within the successful delivery list. The pipeline must make a distinction: redelivery is normal transport behavior; applying a business effect twice is a target correctness bug.

1. Idempotency is a state property

An operation is idempotent for a stated input identity and target contract when repeating it produces the same intended target state as applying it once. Deduplication is one technique: recognize the same event and skip its side effect. A sequence-guarded upsert is another: an older/repeated state cannot overwrite a newer one.

Batch identity answers a different question: which processing attempt/range produced this state? The lab records B20260921-CDC-A for the crashed attempt, B for the successful restart, C for redelivery/reprocessing, and D for the blocked schema event. Because A rolls back, its transactional audit row also rolls back in this simplified harness; a production control plane may retain failed-run telemetry separately.

2. Event identity must detect conflicts, not just duplicates

row = conn.execute(    "SELECT payload_sha256 FROM processed_event WHERE event_id=?",    (e["event_id"],)).fetchone()if row:    if row[0] != payload_hash(e):        raise ValueError("conflicting duplicate event_id")    return "duplicate"

If e202 arrives twice with identical payload, the second delivery is harmless. If an upstream system reuses event_id=e202 for different content, silently ignoring it would hide corruption; the lab fails closed.

3. Sequence-guarded upsert prevents stale overwrite

INSERT INTO fact_sales_current(  order_id,line_no,event_ts,customer_id,product_id,quantity,  extended_amount,extended_cost,status,source_seq) VALUES (?,?,?,?,?,?,?,?,?,?)ON CONFLICT(order_id,line_no) DO UPDATE SET  event_ts=excluded.event_ts,  customer_id=excluded.customer_id,  product_id=excluded.product_id,  quantity=excluded.quantity,  extended_amount=excluded.extended_amount,  extended_cost=excluded.extended_cost,  status=excluded.status,  source_seq=excluded.source_seqWHERE excluded.source_seq > fact_sales_current.source_seq;

This exact syntax is SQLite-specific; MERGE/UPSERT syntax, multiple-match behavior, constraints, isolation, and retry semantics differ across engines. The vendor-neutral contract is “replace current state only with a strictly newer accepted source version for this business key.”

4. Controlled failure: increment a replacement event

-- WRONG if the CDC after-image says quantity is the complete replacement value.UPDATE fact_sales_currentSET quantity = quantity + :after_quantityWHERE order_id=:order_id AND line_no=:line_no;

On redelivery, the quantity increments again. The repair depends on source semantics: if the event is an after-image, assign the replacement value under a newer sequence; if the event is a true business delta, give the delta a stable identity and ensure it is applied at most once. Do not infer delta-vs-replacement from SQL convenience.

5. Deletes need idempotency too

e204 removes O1005/2 from current state and writes one tombstone keyed by delete sequence/event identity. Re-delivering e204 must not fail because the current row is already absent, nor create multiple logical deletes. The local current-state row count drops, while the tombstone proves the deletion was observed.

if current is not None and e["source_seq"] > current[0]:    conn.execute(        "DELETE FROM fact_sales_current WHERE order_id=? AND line_no=?",        (oid, line)    )conn.execute(    "INSERT OR IGNORE INTO sales_tombstone VALUES (?,?,?,?)",    (oid, line, e["source_seq"], e["event_id"]))

6. Retry acceptance evidence

The successful batch receives eight deliveries representing six distinct logical events because e201 and e202 are redelivered. Acceptance output reports applied: 6 and duplicates: 2. A complete rerun after watermark 206 finds no event above the committed cursor and leaves the final SHA-256 unchanged.

That proves convergence for the fixture, not universal “exactly once.” Production tests should include duplicate delivery across process restarts, stale/out-of-order events, conflicting identities, target deadlocks/timeouts, delete replay, and partial downstream publication.

Knowledge check

Summary and next step

This lesson established the mechanism and production boundaries for Idempotent Merge/Upsert Patterns, Batch Identity, Deduplication, and Retry Safety while preserving AtlasMart’s declared grain, governed metrics, history, and reconciliation evidence. Continue to Prove an Incremental Pipeline Handles Restart, Duplicate Delivery, Late Data, Deletes, and Reprocessing with those contracts unchanged unless an explicit, tested migration says otherwise.

Authoritative references

• Python documentation — sqlite3Local DB-API harness used to make commit/rollback and deterministic retry behavior observable.
• SQLite — TransactionTransaction boundaries used by the local acceptance lab; production engines have their own semantics.
• SQLite — UPSERTExact local ON CONFLICT ... DO UPDATE syntax used for sequence-guarded convergence.
• SQLite — DELETELocal current-state delete mechanics; the lab separately preserves a tombstone as CDC evidence.
• Debezium documentationOptional later-course reference for real CDC connector/envelope semantics. Debezium is not required by this chapter or local lab.
• PostgreSQL — Logical DecodingExample of a real database change-stream mechanism; PostgreSQL-specific details are not generalized to all engines.
• Kimball Group — Dimensional Modeling TechniquesDimensional grain/history semantics that incremental loading must preserve.