A daily orders pipeline you can run in DuckDB

1. 01

   Bundled event batch

   Loads one versioned, bundled fixture as a bounded set of input rows; the browser never accepts a visitor-supplied file or URL.

   Inputs

   Approved scenario ID; Sample data from this site, checked against its manifest

   Outputs

   Original event rows; Sample file identity and row count

   SQL queries

   configure_autoload, create_event_ledger

   Quality checks

   No quality check at this stage

   Why, Python, dbt / SQL and schema

   Why

   Decision

   Every batch is a versioned sample file with its hash in a manifest. The run refuses a file whose SHA-256 does not match the manifest, so a result can always be traced to exactly the rows that produced it.

   Alternative considered

   Fetching rows from a URL or letting the reader upload a file.

   Consequence

   Evidence stays reproducible and inspectable, at the cost of only ever running curated data in this version, guarding against a result nobody could reproduce because the input had changed or was never recorded.

   Python

   Reference implementation, Python, not executed

   import hashlibimport jsonfrom pathlib import Pathdef load_batch(path: Path, manifest: dict[str, dict[str, str]]) -> list[dict]:    raw = path.read_bytes()    digest = hashlib.sha256(raw).hexdigest()    if digest != manifest[path.name]["sha256"]:        raise FixtureMismatch(path.name, digest)    rows = json.loads(raw)    if len(rows) > 500:        raise BatchTooLarge(len(rows))    return rows

   dbt / SQL

   Executed in your browser, DuckDB SQL, Bundled scenario data

   The chosen scenario's JSON sample file, loaded by the reviewed runner. Source: lib/features/retail-orders/pipeline.sql.

   SET autoload_known_extensions = false;

   Equivalent dbt model, not executed

   -- models/sources.yml (equivalent declaration, not executed here)version: 2sources:  - name: retail    tables:      - name: events        loaded_at_field: source_updated_at        columns:          - name: event_id            tests: [not_null]

   Schema

   event_id

   VARCHAR, Stable event identity

   order_id

   VARCHAR, Order grain identifier

   source_version

   BIGINT, Version of the order state at the source

   order_date

   DATE, Business date in UTC

   channel

   VARCHAR, web or store

   amount_cents

   BIGINT, Integer minor-unit amount

   source_updated_at

   TIMESTAMP, When the source last changed the order

2. 02

   Event identity gate

   Quarantines missing or empty event IDs first, then rejects a usable ID's conflicting raw payload before any batch mutation.

   Inputs

   Original event IDs and raw payloads; Previously committed event ledger

   Outputs

   Fatal conflict result or rows eligible for validation

   SQL queries

   get_event_ledger, insert_event_ledger

   Quality checks

   No quality check at this stage

   Why, Python, dbt / SQL and schema

   Why

   Decision

   Identity is the event ID plus a SHA-256 of the full raw payload. Same ID and same hash is a safe repeat; same ID and a different hash is a conflict that fails the whole batch before anything is written.

   Alternative considered

   Deduplicating on event ID alone, or keeping the latest payload for a repeated ID.

   Consequence

   A corrupted retry can never hide behind a clean-looking count, and a conflicting batch leaves the committed state untouched, guarding against silently accepting two different facts under one identity.

   Python

   Reference implementation, Python, not executed

   import hashlibimport jsondef payload_hash(row: dict) -> str:    canonical = json.dumps(row, sort_keys=True, separators=(",", ":"))    return hashlib.sha256(canonical.encode()).hexdigest()def gate(rows: list[dict], ledger: dict[str, str]) -> None:    seen: dict[str, str] = {}    for row in rows:        event_id = row.get("event_id")        if not event_id:            continue  # quarantined later with reason missing_event_id        digest = payload_hash(row)        prior = seen.get(event_id) or ledger.get(event_id)        if prior is not None and prior != digest:            raise ConflictingPayload(event_id)  # fail before any mutation        seen[event_id] = digest

   dbt / SQL

   Executed in your browser, DuckDB SQL, Event ledger identity queries

   Reviewed ledger reads and inserts used by the application-owned runner. Source: lib/features/retail-orders/pipeline.sql.

   SELECT event_id, payload_sha256 FROM event_ledger ORDER BY event_id;

   INSERT INTO event_ledger VALUES (?, ?, ?, ?, ?, CAST(? AS DATE), ?, ?);

   Equivalent dbt model, not executed

   -- models/staging/stg_event_identity.sql (equivalent, not executed here)with hashed as (  select    event_id,    md5(to_json(struct_pack(*))) as payload_hash  from {{ source('retail', 'events') }})select event_id, count(distinct payload_hash) as payload_variantsfrom hashedwhere event_id is not nullgroup by 1having count(distinct payload_hash) > 1  -- any row here is a conflict

   Schema

   event_id

   VARCHAR, Identity being checked

   payload_sha256

   VARCHAR, Payload identity used for conflict detection

3. 03

   Validate every original row

   Gives every original row at most one primary quarantine reason; a repeated invalid row is quarantined again, not merged.

   Inputs

   Rows after the ID conflict gate

   Outputs

   Valid candidates; Quarantined rows with primary reasons

   SQL queries

   create_event_ledger

   Quality checks

   No quality check at this stage

   Why, Python, dbt / SQL and schema

   Why

   Decision

   Each original row gets one primary reason, checked in a fixed order, and validation runs before deduplication.

   Alternative considered

   Dropping invalid rows, or collapsing repeats first and validating what is left.

   Consequence

   The quarantine table always reconciles to the input and an invalid repeat is quarantined twice rather than silently merged, so a rejected row can always be explained.

   Python

   Reference implementation, Python, not executed

   REASONS = ("missing_event_id", "missing_order_id", "unknown_channel", "invalid_amount", "invalid_date")def validate(row: dict) -> str | None:    """Return the single primary quarantine reason, or None when the row is valid."""    if not row.get("event_id"):        return "missing_event_id"    if not row.get("order_id"):        return "missing_order_id"    if row.get("channel") not in {"web", "store"}:        return "unknown_channel"    if not isinstance(row.get("amount_cents"), int) or row["amount_cents"] < 0:        return "invalid_amount"    if not is_iso_date(row.get("order_date")):        return "invalid_date"    return None

   dbt / SQL

   Executed in your browser, DuckDB SQL, Validation contract

   The runtime validation contract defines field limits and quarantine reasons. Source: lib/features/retail-orders/pipeline.sql.

   CREATE TABLE IF NOT EXISTS event_ledger (  event_id VARCHAR PRIMARY KEY,  payload_sha256 VARCHAR NOT NULL,  order_id VARCHAR NOT NULL,  source_version BIGINT NOT NULL,  source_updated_at VARCHAR NOT NULL,  order_date DATE NOT NULL,  channel VARCHAR NOT NULL,  amount_cents BIGINT NOT NULL);

   Equivalent dbt model, not executed

   -- models/staging/stg_events_validated.sql (equivalent, not executed here)select  event_id, order_id, source_version, order_date, channel, amount_cents,  case    when event_id is null                    then 'missing_event_id'    when order_id is null                    then 'missing_order_id'    when channel not in ('web', 'store')     then 'unknown_channel'    when amount_cents is null or amount_cents < 0 then 'invalid_amount'    when try_cast(order_date as date) is null then 'invalid_date'  end as quarantine_reasonfrom {{ source('retail', 'events') }}

   Schema

   reason

   ENUM, Primary invalid-row disposition, one per row

   row_number

   BIGINT, Position of the original row in the batch

4. 04

   Collapse valid repeats

   Collapses only raw-identical valid repeats after validation, keeping a within-batch duplicate distinct from a replay of an already-committed event.

   Inputs

   Valid candidates

   Outputs

   Unique valid event candidates; Duplicate disposition count

   SQL queries

   insert_event_ledger

   Quality checks

   No quality check at this stage

   Why, Python, dbt / SQL and schema

   Why

   Decision

   Collapse only rows whose event ID and payload hash are both identical, and count within-batch duplicates separately from replays of already committed events.

   Alternative considered

   Treating every repeated ID as a duplicate, or letting the database's primary key swallow repeats.

   Consequence

   The reconciliation identity holds (input rows equal invalid rows plus duplicates removed plus unique valid events), so counts always add up and a retry is never mistaken for a new order.

   Python

   Reference implementation, Python, not executed

   def dedupe(valid_rows: list[dict]) -> tuple[list[dict], int]:    """Keep the first raw-identical row per event id. Returns (unique_rows, duplicates_removed)."""    unique: dict[str, dict] = {}    duplicates = 0    for row in valid_rows:        key = row["event_id"]        if key in unique:            duplicates += 1  # identical payload guaranteed by the identity gate            continue        unique[key] = row    return list(unique.values()), duplicates

   dbt / SQL

   Executed in your browser, DuckDB SQL, Ledger insert

   The application controls deduplication before the reviewed SQL mutation. Source: lib/features/retail-orders/pipeline.sql.

   INSERT INTO event_ledger VALUES (?, ?, ?, ?, ?, CAST(? AS DATE), ?, ?);

   Equivalent dbt model, not executed

   -- models/intermediate/int_events_unique.sql (equivalent, not executed here)select *from {{ ref('stg_events_validated') }}where quarantine_reason is nullqualify row_number() over (  partition by event_id  order by source_updated_at, event_id) = 1

   Schema

   event_id

   VARCHAR, Deduplication key, unique after this stage

   duplicates_removed

   BIGINT, Identical valid repeats collapsed in this batch

5. 05

   Ledger and current state

   Records every accepted event and the winning order state (by source version then timestamp then event ID) in one transaction; a failed publication rolls the candidate back.

   Inputs

   Unique valid events; Committed event and order state

   Outputs

   Updated event ledger; Winning current state per order

   SQL queries

   begin_transaction, insert_event_ledger, upsert_order_state, commit_transaction, rollback_transaction

   Quality checks

   No quality check at this stage

   Why, Python, dbt / SQL and schema

   Why

   Decision

   Source version wins before timestamp, then the stable event ID breaks ties. The ledger, the state and the facts are staged in one transaction and committed together.

   Alternative considered

   Last-write-wins by arrival time, or updating state and facts in separate steps.

   Consequence

   A late, lower version can never overwrite newer state and replaying a batch produces the same result every time, guarding against out-of-order delivery and half-published runs.

   Python

   Reference implementation, Python, not executed

   from dataclasses import dataclass@dataclass(frozen=True)class OrderState:    order_id: str    source_version: int    source_updated_at: str    event_id: str    order_date: str    channel: str    amount_cents: intdef winner(current: OrderState | None, incoming: OrderState) -> OrderState:    if current is None:        return incoming    rank = lambda s: (s.source_version, s.source_updated_at, s.event_id)    return incoming if rank(incoming) > rank(current) else currentdef apply_batch(conn, unique_rows):    with conn.transaction():          # all or nothing        for row in unique_rows:            conn.insert_ledger(row)            conn.upsert_state(winner(conn.state(row["order_id"]), OrderState(**row)))

   dbt / SQL

   Executed in your browser, DuckDB SQL, State transaction

   Transaction and winner ordering are versioned in the bundled pipeline. Source: lib/features/retail-orders/pipeline.sql.

   BEGIN TRANSACTION;

   INSERT INTO order_state VALUES (?, ?, ?, ?, CAST(? AS DATE), ?, ?)ON CONFLICT (order_id) DO UPDATE SET  source_event_id = excluded.source_event_id,  source_version = excluded.source_version,  source_updated_at = excluded.source_updated_at,  order_date = excluded.order_date,  channel = excluded.channel,  amount_cents = excluded.amount_centsWHERE excluded.source_version > order_state.source_version   OR (excluded.source_version = order_state.source_version AND excluded.source_updated_at > order_state.source_updated_at)   OR (excluded.source_version = order_state.source_version AND excluded.source_updated_at = order_state.source_updated_at AND excluded.source_event_id > order_state.source_event_id);

   COMMIT;

   ROLLBACK;

   Equivalent dbt model, not executed

   -- models/marts/order_state.sql (equivalent, not executed here){{ config(materialized='incremental', unique_key='order_id') }}select *from {{ ref('int_events_unique') }}qualify row_number() over (  partition by order_id  order by source_version desc,           source_updated_at desc,           event_id desc) = 1

   Schema

   source_version

   BIGINT, Primary winner ordering

   source_updated_at

   VARCHAR, Second winner ordering

   source_event_id

   VARCHAR, Deterministic final tie-breaker

6. 06

   One current fact per order

   Materialises one current row per order, replacing the prior date and amount when a correction wins.

   Inputs

   Winning current state per order

   Outputs

   Current order facts; Insert, update and ignored-event counts

   SQL queries

   upsert_order_fact, query_orders_fact

   Quality checks

   State-to-fact reconciliation

   Why, Python, dbt / SQL, schema and checks

   Why

   Decision

   Keep exactly one current row per order, replacing date and amount when a correction wins, rather than appending history to the fact.

   Alternative considered

   An append-only fact with a validity window per version.

   Consequence

   Downstream grouping stays simple and a correction naturally moves an order between days (history lives in the ledger instead), so a changed order is never double counted.

   Python

   Reference implementation, Python, not executed

   def materialize_facts(state_rows):    """One current fact per order; the state winner already resolved corrections."""    return [        {            "order_id": s.order_id,            "source_event_id": s.event_id,            "source_version": s.source_version,            "order_date": s.order_date,            "channel": s.channel,            "amount_cents": s.amount_cents,        }        for s in state_rows    ]

   dbt / SQL

   Executed in your browser, DuckDB SQL, Current order fact

   The reviewed upsert keeps the fact at order grain. Source: lib/features/retail-orders/pipeline.sql.

   INSERT INTO orders_fact VALUES (?, ?, ?, ?, CAST(? AS DATE), ?, ?)ON CONFLICT (order_id) DO UPDATE SET  source_event_id = excluded.source_event_id,  source_version = excluded.source_version,  source_updated_at = excluded.source_updated_at,  order_date = excluded.order_date,  channel = excluded.channel,  amount_cents = excluded.amount_centsWHERE excluded.source_version > orders_fact.source_version   OR (excluded.source_version = orders_fact.source_version AND excluded.source_updated_at > orders_fact.source_updated_at)   OR (excluded.source_version = orders_fact.source_version AND excluded.source_updated_at = orders_fact.source_updated_at AND excluded.source_event_id > orders_fact.source_event_id);

   SELECT order_id, source_event_id, source_version, source_updated_at,       CAST(order_date AS VARCHAR) AS order_date, channel, amount_centsFROM orders_fact ORDER BY order_id;

   Equivalent dbt model, not executed

   -- models/marts/fct_orders.sql (equivalent, not executed here)select  order_id,  event_id       as source_event_id,  source_version,  order_date,  channel,  amount_centsfrom {{ ref('order_state') }}

   Schema

   order_id

   VARCHAR, Primary key at order grain

   order_date

   DATE, UTC reporting date, can change on correction

   channel

   VARCHAR, web or store

   amount_cents

   BIGINT, Current integer amount

   Checks

   • Checked on every runState-to-fact reconciliation
7. 07

   Daily date and channel groups

   Groups current facts by UTC date and channel in integer cents, so a correction removes its old contribution and adds the new one.

   Inputs

   Current order facts

   Outputs

   Order counts and amounts per date/channel group

   SQL queries

   create_daily_mart, query_daily_mart

   Quality checks

   Daily mart reconciliation

   Why, Python, dbt / SQL, schema and checks

   Why

   Decision

   Rebuild the daily mart from current facts on every run instead of incrementing it.

   Alternative considered

   Incremental daily deltas with compensating entries for corrections.

   Consequence

   A correction that moves an order to another day removes its old contribution and adds the new one in the same step, so daily totals never drift from the facts they summarise (the rebuild is cheap at this size).

   Python

   Reference implementation, Python, not executed

   import pandas as pddef daily_mart(facts: pd.DataFrame) -> pd.DataFrame:    mart = (        facts.groupby(["order_date", "channel"], as_index=False)             .agg(orders=("order_id", "count"), amount_cents=("amount_cents", "sum"))    )    assert mart["amount_cents"].sum() == facts["amount_cents"].sum()    return mart

   dbt / SQL

   Executed in your browser, DuckDB SQL, Daily channel mart

   The bundled view derives groups from current facts. Source: lib/features/retail-orders/pipeline.sql.

   CREATE OR REPLACE VIEW daily_channel_mart ASSELECT order_date, channel, COUNT(*) AS order_count, SUM(amount_cents) AS amount_centsFROM orders_factGROUP BY order_date, channel;

   SELECT CAST(order_date AS VARCHAR) AS order_date, channel, order_count, amount_centsFROM daily_channel_mart ORDER BY order_date, channel;

   Equivalent dbt model, not executed

   -- models/marts/daily_orders.sql (equivalent, not executed here)select  order_date,  channel,  count(*)          as orders,  sum(amount_cents) as amount_centsfrom {{ ref('fct_orders') }}group by 1, 2

   Schema

   order_date

   DATE, Grouping date

   channel

   VARCHAR, Grouping channel

   orders

   BIGINT, Count of current facts in the group

   amount_cents

   BIGINT, Group total in cents

   Checks

   • Checked on every runDaily mart reconciliation
8. 08

   Quality checks and publication

   Reconciles state to facts and facts to daily groups, and rejects negative amounts, before publishing the run result.

   Inputs

   Candidate facts and daily groups

   Outputs

   Passed result or a failed run with no partial publication

   SQL queries

   check_state_fact, check_daily_mart, check_nonnegative

   Quality checks

   State-to-fact reconciliation; Daily mart reconciliation; Nonnegative order amounts

   Why, Python, dbt / SQL, schema and checks

   Why

   Decision

   Run reconciliation checks between grains before publication, and treat any failure as a failed run that leaves the previous committed state authoritative.

   Alternative considered

   Publishing first and monitoring afterwards, or checking only row counts.

   Consequence

   A reader can trust that state, facts and daily groups agree at the moment they were published, so a mart can never quietly disagree with the facts or carry a negative or missing amount.

   Python

   Reference implementation, Python, not executed

   def run_checks(conn) -> dict[str, bool]:    results = {        "state_fact_reconciliation": conn.scalar(            "select count(*) from order_state s full join orders_fact f using (order_id) "            "where s.order_id is null or f.order_id is null or s.amount_cents <> f.amount_cents"        ) == 0,        "daily_mart_reconciliation": conn.scalar(            "select (select sum(amount_cents) from daily_mart) = (select sum(amount_cents) from orders_fact)"        ),        "nonnegative_order_amounts": conn.scalar(            "select count(*) from orders_fact where amount_cents < 0"        ) == 0,    }    if not all(results.values()):        raise PublicationBlocked(results)    return results

   dbt / SQL

   Executed in your browser, DuckDB SQL, Quality queries

   The application-owned runner executes these fixed checks after candidate construction. Source: lib/features/retail-orders/pipeline.sql.

   SELECT COUNT(*) AS mismatch_countFROM order_state AS stateFULL OUTER JOIN orders_fact AS fact USING (order_id)WHERE state.source_event_id IS DISTINCT FROM fact.source_event_id   OR state.source_version IS DISTINCT FROM fact.source_version   OR state.source_updated_at IS DISTINCT FROM fact.source_updated_at   OR state.order_date IS DISTINCT FROM fact.order_date   OR state.channel IS DISTINCT FROM fact.channel   OR state.amount_cents IS DISTINCT FROM fact.amount_cents;

   WITH fact_groups AS (  SELECT order_date, channel, COUNT(*) AS order_count, SUM(amount_cents) AS amount_cents  FROM orders_fact GROUP BY order_date, channel), mismatches AS (  (SELECT * FROM fact_groups EXCEPT SELECT * FROM daily_channel_mart)  UNION ALL  (SELECT * FROM daily_channel_mart EXCEPT SELECT * FROM fact_groups))SELECT COUNT(*) AS mismatch_count FROM mismatches;

   SELECT COUNT(*) AS invalid_count FROM orders_fact WHERE amount_cents < 0;

   Equivalent dbt model, not executed

   -- tests/ (equivalent dbt tests, not executed here)-- models/marts/schema.ymlmodels:  - name: fct_orders    columns:      - name: order_id        tests: [unique, not_null]      - name: amount_cents        tests:          - dbt_utils.accepted_range: { min_value: 0 }  - name: daily_orders    tests:      - dbt_utils.equality:          compare_model: ref('fct_orders')          compare_columns: [amount_cents]   # sums reconcile

   Schema

   mismatch_count

   BIGINT, Expected zero for reconciliation checks

   Checks

   • Checked on every runState-to-fact reconciliation
   • Checked on every runDaily mart reconciliation
   • Checked on every runNonnegative order amounts

On a narrow screen, follow the complete ordered list above.