SQL for Data Engineers: Capstones, Whole Pipelines in SQL

-- 1. Conformed dimension first, with the Unknown member
CREATE SEQUENCE product_sk_seq START 1;
CREATE TABLE dim_product (product_sk INTEGER PRIMARY KEY DEFAULT nextval('product_sk_seq'),
                          product_code VARCHAR NOT NULL, product_name VARCHAR, category VARCHAR);
INSERT INTO dim_product VALUES (-1, 'UNKNOWN', 'Unknown', 'Unknown');
INSERT INTO dim_product (product_code, product_name, category)
SELECT product_code, product_name, category FROM raw_product;

-- 2. Transaction facts: resolve keys against the loaded dimension, unknowns to -1
CREATE TABLE fact_sales AS
SELECT s.order_id, s.line_no, COALESCE(d.product_sk, -1) AS product_sk, s.sale_date, s.qty, s.net_amount
FROM raw_sales s LEFT JOIN dim_product d USING (product_code);

CREATE TABLE fact_returns AS
SELECT r.return_id, r.line_no, COALESCE(d.product_sk, -1) AS product_sk, r.return_date, r.qty, r.refund
FROM raw_returns r LEFT JOIN dim_product d USING (product_code);

-- 3. Periodic snapshot fact: one row per product per day
CREATE TABLE fact_inventory_snapshot AS
SELECT COALESCE(d.product_sk, -1) AS product_sk, s.count_date, s.on_hand
FROM raw_stock s LEFT JOIN dim_product d USING (product_code);

-- 4. Prove every grain and every key
SELECT 'fact_sales' AS fact,
       COUNT(*) - COUNT(DISTINCT (order_id, line_no))                          AS duplicate_grain_rows,
       COUNT(*) FILTER (WHERE product_sk NOT IN (SELECT product_sk FROM dim_product)) AS orphan_keys,
       COUNT(*) FILTER (WHERE product_sk = -1)                                 AS unknown_keys
FROM fact_sales
UNION ALL
SELECT 'fact_returns', COUNT(*) - COUNT(DISTINCT (return_id, line_no)),
       COUNT(*) FILTER (WHERE product_sk NOT IN (SELECT product_sk FROM dim_product)),
       COUNT(*) FILTER (WHERE product_sk = -1)
FROM fact_returns
UNION ALL
SELECT 'fact_inventory_snapshot', COUNT(*) - COUNT(DISTINCT (product_sk, count_date)),
       COUNT(*) FILTER (WHERE product_sk NOT IN (SELECT product_sk FROM dim_product)),
       COUNT(*) FILTER (WHERE product_sk = -1)
FROM fact_inventory_snapshot;

-- A. Close current versions that changed, or whose customer is no longer in the snapshot
UPDATE dim_customer AS t
SET is_current = FALSE, effective_to = DATE '2025-06-01' - 1
WHERE t.is_current
  AND NOT EXISTS (SELECT 1 FROM stg_snapshot s
                  WHERE s.customer_code = t.customer_code AND s.row_hash = t.row_hash);

-- B. Open a version for every snapshot customer without a current one:
--    new, changed (just closed in A) and reactivated (closed in an earlier run)
INSERT INTO dim_customer (customer_code, city, effective_from, effective_to, is_current, row_hash, version)
SELECT s.customer_code, s.city, DATE '2025-06-01', DATE '9999-12-31', TRUE, s.row_hash,
       COALESCE((SELECT MAX(version) FROM dim_customer d WHERE d.customer_code = s.customer_code), 0) + 1
FROM stg_snapshot s
WHERE NOT EXISTS (SELECT 1 FROM dim_customer d WHERE d.customer_code = s.customer_code AND d.is_current);

-- 1. Freeze the window and take the batch (deduplicated within itself)
CREATE OR REPLACE TEMP TABLE batch AS
SELECT order_id, line_no, sale_date, net_amount, _loaded_at
FROM src_sales
WHERE _loaded_at >  (SELECT watermark_ts FROM etl_control WHERE table_name = 'fact_sales')
  AND _loaded_at <= (SELECT MAX(_loaded_at) FROM src_sales)
QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id, line_no ORDER BY _loaded_at DESC) = 1;

-- 2. Upsert into the fact on its grain (absorbs redeliveries across batches)
MERGE INTO fact_sales t
USING batch s ON t.order_id = s.order_id AND t.line_no = s.line_no
WHEN MATCHED THEN UPDATE SET sale_date = s.sale_date, net_amount = s.net_amount
WHEN NOT MATCHED THEN INSERT VALUES (s.order_id, s.line_no, s.sale_date, s.net_amount);

-- 3. Recompute only the days this batch touched
DELETE FROM gold_sales_daily WHERE sale_date IN (SELECT DISTINCT sale_date FROM batch);
INSERT INTO gold_sales_daily
SELECT sale_date, SUM(net_amount), COUNT(*)
FROM fact_sales
WHERE sale_date IN (SELECT DISTINCT sale_date FROM batch)
GROUP BY sale_date;

-- 4. Reconcile the recomputed days against the source (must return no rows)
CREATE OR REPLACE TEMP TABLE recon AS
SELECT g.sale_date, g.revenue, x.src_revenue
FROM gold_sales_daily g
JOIN (SELECT sale_date, SUM(net_amount) AS src_revenue
      FROM (SELECT DISTINCT order_id, line_no, sale_date, net_amount FROM src_sales) GROUP BY 1) x USING (sale_date)
WHERE g.sale_date IN (SELECT DISTINCT sale_date FROM batch)
  AND ABS(g.revenue - x.src_revenue) > 0.01;

-- 5. Advance the watermark only if reconciliation passed
UPDATE etl_control
SET watermark_ts = COALESCE((SELECT MAX(_loaded_at) FROM batch), watermark_ts)
WHERE table_name = 'fact_sales'
  AND NOT EXISTS (SELECT 1 FROM recon);

-- 1 + 2. Dedup redeliveries, then build history from EVERY change (post 03, Q39 and Q42)
CREATE OR REPLACE TABLE account_history AS
WITH changes AS (
    SELECT * FROM bronze_cdc
    QUALIFY ROW_NUMBER() OVER (PARTITION BY account_id, seq_no ORDER BY _ingested_at DESC) = 1
),
ordered AS (
    SELECT *, LEAD(source_ts) OVER (PARTITION BY account_id ORDER BY seq_no) AS next_ts
    FROM changes
)
SELECT account_id, balance, status,
       source_ts                                      AS valid_from,
       COALESCE(next_ts, TIMESTAMP '9999-12-31 00:00') AS valid_to
FROM ordered
WHERE op IN ('I', 'U');

-- 3. The state of every account at time T: a half-open window test
SELECT T AS as_of, h.account_id, h.balance, h.status
FROM (VALUES (TIMESTAMP '2025-03-01 13:00'), (TIMESTAMP '2025-03-03 12:00')) AS ts(T)
JOIN account_history h
  ON h.valid_from <= T AND T < h.valid_to
ORDER BY as_of, account_id;

-- Each template is rendered from a dq_check row by the runner (dbt, Airflow, a stored procedure);
-- shown here already rendered for fact_sales.

-- unique: (check 1)
INSERT INTO dq_results
SELECT 1, 'run-42', COUNT(*), COUNT(*) = 0
FROM (SELECT order_id, line_no FROM fact_sales GROUP BY ALL HAVING COUNT(*) > 1);

-- not_null: (check 2)
INSERT INTO dq_results
SELECT 2, 'run-42', COUNT(*), COUNT(*) = 0 FROM fact_sales WHERE net_amount IS NULL;

-- referential: (check 3)
INSERT INTO dq_results
SELECT 3, 'run-42', COUNT(*), COUNT(*) = 0
FROM fact_sales f WHERE NOT EXISTS (SELECT 1 FROM dim_product d WHERE d.product_sk = f.product_sk);

-- control_total: (check 4) compares with a total the source system reported
INSERT INTO dq_results
SELECT 4, 'run-42',
       CASE WHEN ABS(COALESCE(SUM(f.net_amount), 0) - c.expected_total) > 0.01 THEN 1 ELSE 0 END,
       ABS(COALESCE(SUM(f.net_amount), 0) - c.expected_total) <= 0.01
FROM fact_sales f, src_control c WHERE c.table_name = 'fact_sales'
GROUP BY c.expected_total;

-- The publish gate: every check for this run must have passed
SELECT r.check_id, c.check_type, r.rows_failed, r.passed,
       BOOL_AND(r.passed) OVER () AS may_publish
FROM dq_results r JOIN dq_check c USING (check_id)
WHERE r.run_id = 'run-42'
ORDER BY r.check_id;

WITH events AS (                                             -- 1. dedup
    SELECT * FROM raw_events
    QUALIFY ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY event_ts) = 1
),
flagged AS (                                                 -- 2. sessionize
    SELECT *,
           CASE WHEN event_ts - LAG(event_ts) OVER (PARTITION BY user_id ORDER BY event_ts, event_id)
                     <= INTERVAL 30 MINUTE THEN 0 ELSE 1 END AS new_session
    FROM events
),
sessions AS (
    SELECT user_id,
           SUM(new_session) OVER (PARTITION BY user_id ORDER BY event_ts, event_id
                                  ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS session_no,
           event_name, event_ts
    FROM flagged
),
session_facts AS (                                           -- 3. one row per session
    SELECT user_id, session_no,
           CAST(date_trunc('week', MIN(event_ts)) AS DATE) AS week,
           BOOL_OR(event_name = 'purchase')               AS purchased
    FROM sessions GROUP BY user_id, session_no
),
weekly AS (                                                  -- 4. activity per week
    SELECT week,
           COUNT(DISTINCT user_id)                                   AS active_users,
           COUNT(*)                                                  AS sessions,
           ROUND(1.0 * COUNT(*) / COUNT(DISTINCT user_id), 2)        AS sessions_per_user,
           ROUND(100.0 * AVG(CASE WHEN purchased THEN 1 ELSE 0 END), 1) AS purchase_session_pct
    FROM session_facts GROUP BY week
),
cohorts AS (                                                 -- 5. week-1 retention per signup week
    SELECT CAST(date_trunc('week', u.signup_ts) AS DATE) AS week,
           COUNT(DISTINCT u.user_id) AS new_users,
           COUNT(DISTINCT CASE WHEN f.week = CAST(date_trunc('week', u.signup_ts) AS DATE) + 7
                               THEN u.user_id END) AS retained_wk1
    FROM users u LEFT JOIN session_facts f USING (user_id)
    GROUP BY 1
)
SELECT w.week, w.active_users, w.sessions_per_user, w.purchase_session_pct,
       COALESCE(c.new_users, 0)                                         AS new_users,
       ROUND(100.0 * c.retained_wk1 / NULLIF(c.new_users, 0), 0)        AS wk1_retention_pct
FROM weekly w
LEFT JOIN cohorts c USING (week)                                        -- 6. join at the same grain
ORDER BY w.week;