Foundations of distributed data · FND-05

Delivery semantics

3 h
requires FND-01

From your world. Re-running a dbt model safely; SOX 'completeness and accuracy'; idempotent MERGE on a natural key

Make a duplicate-producing pipeline idempotent with a dedup key

The commerce order_items feed deliberately contains exact duplicate events (the same event, delivered again). A naive load double-counts. This is a SOX accuracy failure. You will build the keyed dedup and the idempotent apply that make reprocessing safe.

Functions to implement in `solution.py`

1. `dedup(events, key_fields) -> list[dict]`

events is a list of dicts (rows). key_fields is a list of column names that form the dedup key. Return the list without duplicates. For each distinct key (the tuple of the values of the row at key_fields), keep the first occurrence. Preserve the input order. The result must be the same, however many times the input duplicates each event.

2. `apply_idempotent(state, events, key_fields) -> dict`

state is a dict that maps key_tuple -> row. Apply each event with first-write-wins. Insert row under its key only if that key is not already present. Return a new dict (do not mutate state). Applying the same event again must be a no-op.

3. `revenue_by_product(events) -> list[dict]`

This is a tiny pipeline. First, dedup the raw order_items events on their natural grain ("company_id", "order_item_id"). Then aggregate sum(line_amount) and count(*) per ("company_id", "product_code"). Return rows {"company_id": ..., "product_code": ..., "revenue": float, "n": int}.

What the tests check (DuckDB is the oracle, G6)

dedup collapses the duplicate order-item events of the generator. The row count then equals the COUNT(DISTINCT ...) of DuckDB over the key.
Idempotency property (Hypothesis): dedup(dedup(x)) == dedup(x), and apply_idempotent applied twice equals applied once. This holds for random events, keys, and duplication patterns.
Order-independence: the dedup of a shuffled stream keeps the same set of keys.
revenue_by_product matches an independent DuckDB query over the real commerce data. The query does SELECT ... FROM (SELECT DISTINCT ...) GROUP BY .... A second pass over the events that are already deduped changes nothing.
apply_idempotent does not mutate the input state.

Constraints: pure Python + duckdb (oracle only, in the tests). Deterministic. lab: none.

Starter code

solution.py

"""Starter for FND-05/01-idempotent-pipeline. See README.md."""

from __future__ import annotations


def dedup(events, key_fields) -> list[dict]:
    # TODO: keep the first row per key tuple, preserving order.
    raise NotImplementedError


def apply_idempotent(state, events, key_fields) -> dict:
    # TODO: first-write-wins upsert into a NEW dict keyed by the dedup key.
    raise NotImplementedError


def revenue_by_product(events) -> list[dict]:
    # TODO: dedup on (company_id, order_item_id), then sum/count per
    #       (company_id, product_code).
    raise NotImplementedError

Hints

Hint 1

dedup: build the dedup key for each row as the tuple of key_fields values. Keep the FIRST row seen for each key (track a set of seen keys), preserving input order. Duplicate events (same key) collapse to one.

Hint 2

apply_idempotent: treat state as a dict keyed by the dedup key. Applying an event sets state[key] = row only if the key is not already present (first-write-wins), so re-applying the same event is a no-op. Return a NEW state dict; don't mutate the input.

Hint 3

For order_items the natural grain is (company_id, order_item_id). The generator emits exact duplicate order-item events on purpose; a correct key-based dedup makes the row count match DuckDB's COUNT(DISTINCT ...) and makes a second pass change nothing (idempotent).

Run it locally

uv run de start FND-05/01-idempotent-pipeline
# edit workspace/FND-05/01-idempotent-pipeline/
uv run de test