Requirements & grain:- Grain: one row per order (order_id) representing the lifecycle for that order. Use accumulating snapshot to store final timestamps for each stage and measure elapsed times.- Constraints: stages: order_placed → picked → packed → shipped → delivered. Stages may arrive out-of-order, retry, or rollback.Recommended schema:- order_id (PK)- customer_id- order_placed_at TIMESTAMP NOT NULL- picked_at TIMESTAMP NULL- packed_at TIMESTAMP NULL- shipped_at TIMESTAMP NULL- delivered_at TIMESTAMP NULL- current_status VARCHAR- attempts JSONB (optional) — store retries metadata: [{"stage":"picked","ts":"...","attempt":2},...]- last_updated_at TIMESTAMP- is_closed BOOLEANMeasures (derived / materialized):- time_to_pick = EXTRACT(EPOCH FROM picked_at - order_placed_at)- time_to_pack = EXTRACT(EPOCH FROM packed_at - picked_at)- time_to_ship = EXTRACT(EPOCH FROM shipped_at - packed_at)- time_placed_to_shipped = EXTRACT(EPOCH FROM shipped_at - order_placed_at)Handling updates & edge cases:- Idempotent, event-driven upserts using source event's event_id and event_ts.- Use an ETL/CDC job that applies business rules: - Only set stage timestamp if incoming_ts is non-null and (existing_ts IS NULL OR incoming_ts < existing_ts for earliest-timestamp semantics, or incoming_ts > existing_ts for latest-timestamp semantics). Pick a rule (recommend earliest for start stages, latest for completion). - For retries: append to attempts JSONB and optionally update stage timestamp to latest successful attempt. - For rollbacks: accept explicit rollback events that clear downstream timestamps (e.g., rollback_to_picked clears packed_at, shipped_at, delivered_at) and set current_status. - Use last_event_id to guarantee exactly-once ordering; apply out-of-order events by comparing event_ts and using a conflict resolution policy.- Maintain audit/log table of raw events for replay and debugging.Example queries:Average time placed → shipped (seconds):sql
SELECT
AVG(EXTRACT(EPOCH FROM shipped_at - order_placed_at)) AS avg_seconds,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM shipped_at - order_placed_at)) AS median_seconds,
COUNT(*) AS orders_count
FROM accumulating_orders
WHERE shipped_at IS NOT NULL
AND order_placed_at IS NOT NULL;
sql
SELECT
DATE(order_placed_at) AS day,
AVG(EXTRACT(EPOCH FROM shipped_at - order_placed_at))/3600 AS avg_hours,
COUNT(*) AS n
FROM accumulating_orders
WHERE shipped_at IS NOT NULL
GROUP BY 1
ORDER BY 1;
