資深資料工程師 SQL 面試題

-- PostgreSQL
CREATE TABLE customers (
  customer_id  INT PRIMARY KEY,
  name         TEXT NOT NULL,
  country      TEXT
);

CREATE TABLE orders (
  order_id     INT PRIMARY KEY,
  customer_id  INT,                     -- NULL for a guest checkout
  order_date   DATE NOT NULL,
  amount       NUMERIC(10, 2) NOT NULL,
  status       TEXT NOT NULL            -- 'paid', 'refunded'
);

CREATE TABLE events (
  event_id     INT NOT NULL,            -- the same event can arrive twice
  user_id      INT NOT NULL,
  event_type   TEXT NOT NULL,           -- 'visit', 'purchase'
  event_time   TIMESTAMP NOT NULL,
  updated_at   TIMESTAMP NOT NULL,
  ingest_seq   BIGINT NOT NULL          -- monotonic, assigned on arrival
);

CREATE TABLE employees (
  employee_id   INT PRIMARY KEY,
  name          TEXT NOT NULL,
  department_id INT NOT NULL,
  manager_id    INT,                    -- NULL for the CEO
  salary        NUMERIC(10, 2) NOT NULL
);

CREATE TABLE daily_revenue (order_date DATE PRIMARY KEY, revenue NUMERIC(12, 2) NOT NULL);
CREATE TABLE user_logins   (user_id INT NOT NULL, login_date DATE NOT NULL);

INSERT INTO customers VALUES (1, 'Ann', 'TW'), (2, 'Bo', 'JP'), (3, 'Cy', 'TW'), (4, 'Di', 'US');

INSERT INTO orders VALUES
  (101, 1,    DATE '2026-11-01', 30, 'paid'),
  (102, 1,    DATE '2026-11-02', 20, 'paid'),
  (103, 2,    DATE '2026-11-02', 50, 'refunded'),
  (104, NULL, DATE '2026-11-03', 10, 'paid');

INSERT INTO events VALUES
  (1, 10, 'visit',    TIMESTAMP '2026-11-01 09:00', TIMESTAMP '2026-11-01 09:00', 1),
  (1, 10, 'visit',    TIMESTAMP '2026-11-01 09:00', TIMESTAMP '2026-11-01 09:05', 2),
  (2, 10, 'purchase', TIMESTAMP '2026-11-01 09:20', TIMESTAMP '2026-11-01 09:20', 3),
  (3, 11, 'visit',    TIMESTAMP '2026-11-01 10:00', TIMESTAMP '2026-11-01 10:00', 4),
  (4, 10, 'visit',    TIMESTAMP '2026-11-01 11:00', TIMESTAMP '2026-11-01 11:00', 5),
  (5, 12, 'purchase', TIMESTAMP '2026-11-02 08:00', TIMESTAMP '2026-11-02 08:00', 6),
  (6, 11, 'visit',    TIMESTAMP '2026-11-02 09:00', TIMESTAMP '2026-11-02 09:00', 7);

INSERT INTO employees VALUES
  (1, 'Kai', 1, NULL, 300), (2, 'Lin', 1, 1, 200), (3, 'Max', 1, 1, 200),
  (4, 'Ned', 1, 2,    150), (5, 'Oli', 1, 2, 120), (6, 'Pat', 2, 1, 180),
  (7, 'Quy', 2, 6,     90);

INSERT INTO daily_revenue VALUES               -- 11-04 and 11-07 are missing
  (DATE '2026-11-01', 100), (DATE '2026-11-02', 120), (DATE '2026-11-03', 90),
  (DATE '2026-11-05', 110), (DATE '2026-11-06', 130), (DATE '2026-11-08', 70),
  (DATE '2026-11-09', 150), (DATE '2026-11-10', 140);

INSERT INTO user_logins VALUES                 -- user 10 logged in twice on 11-02
  (10, DATE '2026-11-01'), (10, DATE '2026-11-02'), (10, DATE '2026-11-02'),
  (10, DATE '2026-11-03'), (10, DATE '2026-11-05'),
  (11, DATE '2026-11-01'), (11, DATE '2026-11-03'), (11, DATE '2026-11-04'),
  (11, DATE '2026-11-05'), (11, DATE '2026-11-06');

-- aggregate first, join second
WITH order_totals AS (
  SELECT customer_id, SUM(amount) AS total_amount, COUNT(*) AS order_count
  FROM   orders
  WHERE  customer_id IS NOT NULL
  GROUP  BY customer_id
)
SELECT c.customer_id, c.name,
       COALESCE(t.total_amount, 0) AS total_amount,
       COALESCE(t.order_count, 0)  AS order_count
FROM   customers c
LEFT   JOIN order_totals t ON t.customer_id = c.customer_id;

-- the key distribution: is a handful of keys carrying most of the rows?
SELECT customer_id, COUNT(*) AS n
FROM   orders
GROUP  BY customer_id
ORDER  BY n DESC
LIMIT  20;

-- Spark SQL
SELECT /*+ BROADCAST(c) */ c.customer_id, SUM(o.amount) AS total_amount
FROM   orders o
JOIN   customers c ON c.customer_id = o.customer_id
WHERE  c.country = 'TW'
GROUP  BY c.customer_id;

-- the assumption: one row per customer_id on the right
SELECT customer_id, COUNT(*) FROM customers GROUP BY customer_id HAVING COUNT(*) > 1;

-- the reconciliation: the join must not change the total
SELECT (SELECT SUM(amount) FROM orders WHERE customer_id IS NOT NULL) AS before_join,
       (SELECT SUM(o.amount) FROM orders o JOIN customers c USING (customer_id)) AS after_join;

-- NOT EXISTS: the default
SELECT c.customer_id
FROM   customers c
WHERE  NOT EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.customer_id);

-- LEFT JOIN ... IS NULL: the same result
SELECT c.customer_id
FROM   customers c
LEFT   JOIN orders o ON o.customer_id = c.customer_id
WHERE  o.customer_id IS NULL;

-- NOT IN: returns NO rows, because orders.customer_id contains a NULL
SELECT c.customer_id
FROM   customers c
WHERE  c.customer_id NOT IN (SELECT customer_id FROM orders);

-- many-to-many made one-to-many: reduce the right side to one row per key first
WITH latest_status AS (
  SELECT DISTINCT ON (customer_id) customer_id, status
  FROM   orders
  WHERE  customer_id IS NOT NULL
  ORDER  BY customer_id, order_date DESC, order_id DESC
)
SELECT c.customer_id, s.status
FROM   customers c
LEFT   JOIN latest_status s USING (customer_id);

WITH ranked AS (
  SELECT *,
         ROW_NUMBER() OVER (
           PARTITION BY event_id
           ORDER BY updated_at DESC, ingest_seq DESC   -- the tie-breaker
         ) AS rn
  FROM   events
)
SELECT event_id, user_id, event_type, event_time, updated_at
FROM   ranked
WHERE  rn = 1;

-- DuckDB (Spark: max_by(struct(*), struct(updated_at, ingest_seq)))
SELECT event_id,
       arg_max(user_id,    (updated_at, ingest_seq)) AS user_id,
       arg_max(event_type, (updated_at, ingest_seq)) AS event_type
FROM   events
GROUP  BY event_id;

CREATE TABLE events_clean AS
SELECT event_id, user_id, event_type, event_time, updated_at, ingest_seq
FROM   events WHERE FALSE;                          -- an empty copy, unique on event_id

MERGE INTO events_clean t
USING (
  SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY event_id
                                 ORDER BY updated_at DESC, ingest_seq DESC) AS rn
    FROM   events
    WHERE  ingest_seq > 0                            -- the new batch: above the last watermark
  ) b WHERE rn = 1
) s
ON t.event_id = s.event_id
WHEN MATCHED AND (s.updated_at, s.ingest_seq) > (t.updated_at, t.ingest_seq) THEN
  UPDATE SET user_id = s.user_id, event_type = s.event_type, event_time = s.event_time,
             updated_at = s.updated_at, ingest_seq = s.ingest_seq
WHEN NOT MATCHED THEN
  INSERT (event_id, user_id, event_type, event_time, updated_at, ingest_seq)
  VALUES (s.event_id, s.user_id, s.event_type, s.event_time, s.updated_at, s.ingest_seq);

CREATE TABLE raw_orders (order_id INT, status TEXT, amount NUMERIC(10, 2), updated_at TIMESTAMP);
INSERT INTO raw_orders VALUES
  (1, 'PENDING', 30, TIMESTAMP '2026-11-01 10:00'),
  (1, 'SUCCESS', 30, TIMESTAMP '2026-11-01 10:01'),
  (1, 'FAILED',  30, TIMESTAMP '2026-11-01 10:05'),   -- a late retry failure: must not win
  (2, 'FAILED',  50, TIMESTAMP '2026-11-01 11:00'),
  (2, 'PENDING', 50, TIMESTAMP '2026-11-01 11:02');

WITH prioritized AS (
  SELECT *,
         ROW_NUMBER() OVER (
           PARTITION BY order_id
           ORDER BY CASE status WHEN 'SUCCESS' THEN 1
                                WHEN 'PENDING' THEN 2
                                ELSE 3 END,
                    updated_at DESC
         ) AS rn
  FROM   raw_orders
)
SELECT order_id, status, amount, updated_at FROM prioritized WHERE rn = 1;

WITH ranked AS (
  SELECT department_id, employee_id, name, salary,
         DENSE_RANK() OVER (PARTITION BY department_id ORDER BY salary DESC) AS salary_rank
  FROM   employees
)
SELECT department_id, employee_id, name, salary
FROM   ranked
WHERE  salary_rank = 2;

SELECT department_id, name, salary
FROM (
  SELECT department_id, name, salary,
         DENSE_RANK() OVER (PARTITION BY department_id ORDER BY salary DESC) AS r
  FROM   employees
) t
WHERE  r <= 3
ORDER  BY department_id, salary DESC;

-- the version usually written first: correct only if every date has a row
SELECT order_date, revenue,
       AVG(revenue) OVER (ORDER BY order_date
                          ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS rolling_avg_rows
FROM   daily_revenue;

WITH calendar AS (
  SELECT CAST(d AS DATE) AS order_date
  FROM   generate_series(DATE '2026-11-01', DATE '2026-11-10', INTERVAL '1 day') AS g(d)
),
filled AS (
  SELECT c.order_date, COALESCE(r.revenue, 0) AS revenue    -- a missing day counts as zero
  FROM   calendar c
  LEFT   JOIN daily_revenue r ON r.order_date = c.order_date
)
SELECT order_date, revenue,
       AVG(revenue) OVER w AS rolling_avg_7d,
       COUNT(*)     OVER w AS days_in_window       -- < 7 for the first six days
FROM   filled
WINDOW w AS (ORDER BY order_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW);

SELECT order_date, revenue,
       AVG(revenue) OVER (ORDER BY order_date
                          RANGE BETWEEN INTERVAL '6 days' PRECEDING AND CURRENT ROW)
         AS rolling_avg_days_present
FROM   daily_revenue;

SELECT CAST(event_time AS DATE) AS event_date,
       COUNT(DISTINCT CASE WHEN event_type = 'visit'    THEN user_id END) AS visitors,
       COUNT(DISTINCT CASE WHEN event_type = 'purchase' THEN user_id END) AS purchasers,
       1.0 * COUNT(DISTINCT CASE WHEN event_type = 'purchase' THEN user_id END)
           / NULLIF(COUNT(DISTINCT CASE WHEN event_type = 'visit' THEN user_id END), 0)
         AS conversion_rate
FROM   events
GROUP  BY CAST(event_time AS DATE)
ORDER  BY event_date;

WITH user_daily AS (
  SELECT CAST(event_time AS DATE) AS event_date, user_id,
         MAX(CASE WHEN event_type = 'visit'    THEN 1 ELSE 0 END) AS visited,
         MAX(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchased
  FROM   events
  GROUP  BY CAST(event_time AS DATE), user_id
)
SELECT event_date,
       SUM(visited)                                              AS visitors,
       SUM(CASE WHEN visited = 1 AND purchased = 1 THEN 1 ELSE 0 END) AS converted,
       1.0 * SUM(CASE WHEN visited = 1 AND purchased = 1 THEN 1 ELSE 0 END)
           / NULLIF(SUM(visited), 0)                             AS conversion_rate
FROM   user_daily
GROUP  BY event_date
ORDER  BY event_date;

WITH visits AS (
  SELECT user_id, CAST(event_time AS DATE) AS visit_date, MIN(event_time) AS first_visit
  FROM   events WHERE event_type = 'visit'
  GROUP  BY user_id, CAST(event_time AS DATE)
)
SELECT v.visit_date,
       COUNT(*)                     AS visitors,
       COUNT(p.user_id)             AS converted,
       1.0 * COUNT(p.user_id) / COUNT(*) AS conversion_rate
FROM   visits v
LEFT   JOIN LATERAL (
  SELECT e.user_id FROM events e
  WHERE  e.user_id = v.user_id AND e.event_type = 'purchase'
    AND  e.event_time >= v.first_visit
    AND  e.event_time <  v.first_visit + INTERVAL '24 hours'
  LIMIT  1
) p ON TRUE
GROUP  BY v.visit_date
ORDER  BY v.visit_date;

CREATE TABLE users (user_id INT PRIMARY KEY, signup_ts TIMESTAMP NOT NULL);
INSERT INTO users VALUES
  (10, TIMESTAMP '2026-11-01 08:00'), (11, TIMESTAMP '2026-11-01 09:30'),
  (12, TIMESTAMP '2026-11-02 07:00'), (13, TIMESTAMP '2026-11-01 12:00');

WITH cohort AS (
  SELECT user_id, CAST(signup_ts AS DATE) AS cohort_date FROM users
),
cohort_size AS (
  SELECT cohort_date, COUNT(*) AS cohort_users FROM cohort GROUP BY cohort_date
),
activity AS (
  SELECT DISTINCT user_id, CAST(event_time AS DATE) AS activity_date FROM events
),
retained AS (
  SELECT c.cohort_date,
         a.activity_date - c.cohort_date AS day_number,
         COUNT(DISTINCT a.user_id)        AS active_users
  FROM   cohort c
  JOIN   activity a ON a.user_id = c.user_id AND a.activity_date >= c.cohort_date
  GROUP  BY c.cohort_date, a.activity_date - c.cohort_date
),
offsets AS (                                   -- the day offsets the report asks for
  SELECT day_number FROM (VALUES (0), (1), (7), (30)) AS o(day_number)
),
eligible AS (                                  -- only offsets a cohort is old enough to have
  SELECT s.cohort_date, s.cohort_users, o.day_number
  FROM   cohort_size s
  CROSS  JOIN offsets o
  WHERE  s.cohort_date + o.day_number <= (SELECT MAX(activity_date) FROM activity)
)
SELECT e.cohort_date, e.day_number, e.cohort_users,
       COALESCE(r.active_users, 0)                                     AS active_users,
       ROUND(100.0 * COALESCE(r.active_users, 0) / e.cohort_users, 1) AS retention_pct
FROM   eligible e
LEFT   JOIN retained r ON r.cohort_date = e.cohort_date AND r.day_number = e.day_number
ORDER  BY e.cohort_date, e.day_number;

WITH user_days AS (
  SELECT DISTINCT user_id, CAST(event_time AS DATE) AS d FROM events
),
calendar AS (
  SELECT DISTINCT d FROM user_days
)
SELECT c.d,
       COUNT(DISTINCT CASE WHEN u.d = c.d                       THEN u.user_id END) AS dau,
       COUNT(DISTINCT CASE WHEN u.d >  c.d - 7                  THEN u.user_id END) AS wau,
       COUNT(DISTINCT u.user_id)                                                    AS mau
FROM   calendar c
JOIN   user_days u ON u.d > c.d - 30 AND u.d <= c.d       -- a trailing 30-day window
GROUP  BY c.d
ORDER  BY c.d;

SELECT customer_id,
       MIN(order_date) AS first_order_date,
       MAX(order_date) AS last_order_date,
       COUNT(*)        AS orders,
       MAX(order_date) - MIN(order_date) AS active_span_days
FROM   orders
WHERE  status = 'paid' AND customer_id IS NOT NULL
GROUP  BY customer_id;

SELECT customer_id,
       SUM(CASE WHEN status = 'paid'     THEN amount ELSE 0 END) AS paid_amount,
       SUM(CASE WHEN status = 'refunded' THEN amount ELSE 0 END) AS refunded_amount,
       COUNT(*) FILTER (WHERE status = 'paid')                   AS paid_orders
FROM   orders
WHERE  order_date >= DATE '2026-11-01' AND order_date < DATE '2026-12-01'
  AND  customer_id IS NOT NULL
GROUP  BY customer_id;

-- one row per month, one column per status
SELECT DATE_TRUNC('month', order_date) AS month,
       SUM(amount) FILTER (WHERE status = 'paid')     AS paid,
       SUM(amount) FILTER (WHERE status = 'refunded') AS refunded
FROM   orders
GROUP  BY DATE_TRUNC('month', order_date)
ORDER  BY month;

WITH login_days AS (
  SELECT DISTINCT user_id, login_date FROM user_logins      -- one row per user per day
),
numbered AS (
  SELECT user_id, login_date,
         login_date - CAST(ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY login_date) AS INT)
           AS island_id
  FROM   login_days
)
SELECT user_id,
       MIN(login_date) AS streak_start,
       MAX(login_date) AS streak_end,
       COUNT(*)        AS streak_days
FROM   numbered
GROUP  BY user_id, island_id
HAVING COUNT(*) >= 3
ORDER  BY user_id, streak_start;

user 11   login_date   row_number   login_date - row_number
          11-01        1            10-31   island A
          11-03        2            11-01   island B
          11-04        3            11-01   island B
          11-05        4            11-01   island B
          11-06        5            11-01   island B   -> 4 days

WITH ordered AS (
  SELECT user_id, event_time,
         LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time) AS prev_time
  FROM   events
),
flagged AS (
  SELECT user_id, event_time,
         CASE WHEN prev_time IS NULL
                OR event_time > prev_time + INTERVAL '30 minutes' THEN 1 ELSE 0 END
           AS new_session
  FROM   ordered
)
SELECT user_id, event_time,
       SUM(new_session) OVER (PARTITION BY user_id ORDER BY event_time
                              ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS session_no
FROM   flagged
ORDER  BY user_id, event_time;

WITH calendar AS (
  SELECT CAST(d AS DATE) AS dt
  FROM   generate_series(DATE '2026-11-01', DATE '2026-11-10', INTERVAL '1 day') AS g(d)
)
SELECT c.dt AS missing_date
FROM   calendar c
WHERE  NOT EXISTS (SELECT 1 FROM daily_revenue r WHERE r.order_date = c.dt)
ORDER  BY c.dt;

-- returns rows only when something is wrong; the orchestrator fails the task if any come back
WITH calendar AS (
  SELECT CAST(d AS DATE) AS dt
  FROM   generate_series(DATE '2026-11-01', DATE '2026-11-10', INTERVAL '1 day') AS g(d)
)
SELECT 'missing_date' AS check_name, CAST(dt AS TEXT) AS detail
FROM   calendar c
WHERE  NOT EXISTS (SELECT 1 FROM daily_revenue r WHERE r.order_date = c.dt)
UNION ALL
SELECT 'negative_revenue', CAST(order_date AS TEXT)
FROM   daily_revenue WHERE revenue < 0;

WITH RECURSIVE reports AS (
  -- anchor: the starting manager
  SELECT employee_id, manager_id, name, 0 AS depth, ARRAY[employee_id] AS path
  FROM   employees
  WHERE  employee_id = 2

  UNION ALL

  -- recursive member: the direct reports of everyone found so far
  SELECT e.employee_id, e.manager_id, e.name, r.depth + 1, r.path || ARRAY[e.employee_id]
  FROM   employees e
  JOIN   reports  r ON e.manager_id = r.employee_id
  WHERE  e.employee_id <> ALL (r.path)          -- cycle guard: never revisit a node
    AND  r.depth < 20                           -- depth guard: a backstop
)
SELECT employee_id, name, depth, path
FROM   reports
WHERE  depth > 0
ORDER  BY depth, employee_id;

-- prunes: a range on the raw column
SELECT SUM(amount) FROM orders
WHERE  order_date >= DATE '2026-11-01' AND order_date < DATE '2026-12-01';

-- does not prune on most engines: the column is wrapped in a function
SELECT SUM(amount) FROM orders
WHERE  EXTRACT(YEAR FROM order_date) = 2026 AND EXTRACT(MONTH FROM order_date) = 11;

-- Spark SQL: salt the big side, explode the small side
WITH salted_orders AS (
  SELECT *, CAST(FLOOR(RAND() * 8) AS INT) AS salt FROM orders
),
salted_customers AS (
  SELECT c.*, s.salt FROM customers c
  CROSS JOIN (SELECT EXPLODE(SEQUENCE(0, 7)) AS salt) s
)
SELECT o.customer_id, SUM(o.amount) AS total_amount
FROM   salted_orders o
JOIN   salted_customers c ON c.customer_id = o.customer_id AND c.salt = o.salt
GROUP  BY o.customer_id;

-- anti-pattern: a per-row subquery in the SELECT list
SELECT c.customer_id,
       (SELECT SUM(o.amount) FROM orders o WHERE o.customer_id = c.customer_id) AS total
FROM   customers c;

-- rewrite: one aggregate, one join
SELECT c.customer_id, t.total
FROM   customers c
LEFT   JOIN (SELECT customer_id, SUM(amount) AS total
             FROM orders GROUP BY customer_id) t USING (customer_id);

-- spark sql — prove it: no Exchange on either branch, from the join down through the Sort to the scan
EXPLAIN FORMATTED
SELECT c.customer_id, c.customer_name, SUM(o.amount) AS total_amount
FROM   customers_bucketed c
JOIN   orders_bucketed    o ON c.customer_id = o.customer_id
GROUP  BY c.customer_id, c.customer_name;

CREATE TABLE etl_watermark (job TEXT PRIMARY KEY, high_water TIMESTAMP NOT NULL);
INSERT INTO etl_watermark VALUES ('orders_sync', TIMESTAMP '2026-11-01 00:00');

CREATE TABLE src_orders (order_id INT, status TEXT, amount NUMERIC(10, 2),
                         updated_at TIMESTAMP NOT NULL,
                         change_seq BIGINT    NOT NULL);   -- change_seq: the source's order
INSERT INTO src_orders VALUES
  (1, 'paid',     30, TIMESTAMP '2026-11-01 10:00', 1),
  (1, 'refunded', 30, TIMESTAMP '2026-11-01 15:00', 3),  -- two versions of one key in the batch
  (2, 'paid',     50, TIMESTAMP '2026-11-01 11:00', 2),
  (2, 'paid',     55, TIMESTAMP '2026-11-01 11:00', 4);  -- a correction with the same timestamp

CREATE TABLE dim_orders (order_id INT PRIMARY KEY, status TEXT, amount NUMERIC(10, 2),
                         updated_at TIMESTAMP NOT NULL, change_seq BIGINT NOT NULL);

BEGIN;

-- 1. snapshot the batch once: above the watermark (minus a lookback), one version per key
DROP TABLE IF EXISTS stg_orders;                       -- safe to rerun in the same session
CREATE TEMP TABLE stg_orders AS
SELECT order_id, status, amount, updated_at, change_seq
FROM (
  SELECT s.*, ROW_NUMBER() OVER (PARTITION BY order_id
                                 ORDER BY updated_at DESC, change_seq DESC) AS rn
  FROM   src_orders s
  WHERE  s.updated_at > (SELECT high_water - INTERVAL '1 hour'
                         FROM etl_watermark WHERE job = 'orders_sync')
) b
WHERE rn = 1;

-- 2. replace only with a newer version, so replaying an old batch changes nothing
MERGE INTO dim_orders t
USING stg_orders s
ON t.order_id = s.order_id
WHEN MATCHED AND (s.updated_at, s.change_seq) > (t.updated_at, t.change_seq) THEN
  UPDATE SET status = s.status, amount = s.amount,
             updated_at = s.updated_at, change_seq = s.change_seq
WHEN NOT MATCHED THEN
  INSERT (order_id, status, amount, updated_at, change_seq)
  VALUES (s.order_id, s.status, s.amount, s.updated_at, s.change_seq);

-- 3. advance the watermark to the newest row of the batch that was merged
UPDATE etl_watermark
SET    high_water = GREATEST(high_water,
                             COALESCE((SELECT MAX(updated_at) FROM stg_orders), high_water))
WHERE  job = 'orders_sync';                            -- never moves backwards

DROP TABLE stg_orders;
COMMIT;

source table / CDC stream
  -> staging table              the raw batch, tagged with a batch_id
  -> dedupe + validate          one version per key; reject rows that fail checks
  -> MERGE INTO target          idempotent: a replayed batch rewrites the same rows
  -> commit watermark + audit   only now advance the watermark; record batch_id,
                                rows read / inserted / updated, processing time

CREATE TABLE dim_customer (
  customer_sk  INT PRIMARY KEY,
  customer_id  INT NOT NULL,
  address      TEXT NOT NULL,
  valid_from   DATE NOT NULL,
  valid_to     DATE,                    -- NULL while current; exclusive end otherwise
  is_current   BOOLEAN NOT NULL
);
INSERT INTO dim_customer VALUES
  (1, 101, 'Taipei', DATE '2025-01-01', NULL, TRUE),
  (2, 102, 'Osaka',  DATE '2025-03-01', NULL, TRUE);

INSERT INTO dim_customer VALUES
  (3, 104, 'Busan',  DATE '2025-05-01', NULL, TRUE);

CREATE TABLE stg_customer (customer_id INT, address TEXT, change_date DATE, change_seq BIGINT);
INSERT INTO stg_customer VALUES
  (101, 'Kyoto',   DATE '2025-07-01', 1),  -- moved twice on one day:
  (101, 'Tokyo',   DATE '2025-07-01', 2),  --   the day's last change wins
  (101, 'Nagoya',  DATE '2025-07-03', 3),  -- and again two days later: a version of its own
  (102, 'Osaka',   DATE '2025-07-01', 4),  -- unchanged
  (103, 'Seoul',   DATE '2025-07-01', 5),  -- new customer
  (104, 'Incheon', DATE '2025-04-01', 6);  -- late: older than the current Busan row

-- changes too old to apply here wait for the history rebuild (kept across runs)
CREATE TABLE dim_customer_late (customer_id INT, address TEXT, change_date DATE);

BEGIN;
DROP TABLE IF EXISTS stg_daily;                -- safe to rerun in the same session
DROP TABLE IF EXISTS stg_chain;

-- 0. one change per customer per day: the day's last
CREATE TEMP TABLE stg_daily AS
SELECT customer_id, address, change_date
FROM (
  SELECT s.*, ROW_NUMBER() OVER (PARTITION BY customer_id, change_date
                                 ORDER BY change_seq DESC) AS rn
  FROM   stg_customer s
) t
WHERE rn = 1;

-- 1. route late changes: dated on or before the current row's start and different from the
--    version valid on that day (a replayed change matches it and is skipped)
INSERT INTO dim_customer_late
SELECT s.customer_id, s.address, s.change_date
FROM   stg_daily s
JOIN   dim_customer cur ON cur.customer_id = s.customer_id AND cur.is_current
LEFT   JOIN dim_customer h ON h.customer_id = s.customer_id
                          AND h.valid_from <= s.change_date
                          AND (s.change_date < h.valid_to OR h.valid_to IS NULL)
WHERE  s.change_date <= cur.valid_from
  AND  h.address IS DISTINCT FROM s.address
  AND  NOT EXISTS (SELECT 1 FROM dim_customer_late l
                   WHERE l.customer_id = s.customer_id AND l.change_date = s.change_date
                     AND l.address = s.address);

-- 2. the new versions: changes after the current row's start, minus no-ops, each one ending
--    where the next begins
CREATE TEMP TABLE stg_chain AS
WITH newer AS (
  SELECT s.customer_id, s.address, s.change_date,
         COALESCE(LAG(s.address) OVER (PARTITION BY s.customer_id ORDER BY s.change_date),
                  cur.address) AS prev_address
  FROM   stg_daily s
  LEFT   JOIN dim_customer cur ON cur.customer_id = s.customer_id AND cur.is_current
  WHERE  cur.customer_id IS NULL OR s.change_date > cur.valid_from
),
changed AS (
  SELECT * FROM newer WHERE address IS DISTINCT FROM prev_address
)
SELECT customer_id, address, change_date AS valid_from,
       LEAD(change_date) OVER (PARTITION BY customer_id ORDER BY change_date) AS valid_to
FROM   changed;

-- 3. close each changed customer's current row where its first new version starts
UPDATE dim_customer d
SET    valid_to = c.first_from, is_current = FALSE
FROM   (SELECT customer_id, MIN(valid_from) AS first_from
        FROM stg_chain GROUP BY customer_id) c
WHERE  d.customer_id = c.customer_id
  AND  d.is_current;

-- 4. insert the chain; only its last version is current
INSERT INTO dim_customer
SELECT (SELECT COALESCE(MAX(customer_sk), 0) FROM dim_customer)
         + ROW_NUMBER() OVER (ORDER BY customer_id, valid_from),
       customer_id, address, valid_from, valid_to, valid_to IS NULL
FROM   stg_chain;

DROP TABLE stg_daily;
DROP TABLE stg_chain;
COMMIT;

-- PostgreSQL only: a partial unique index (DuckDB, Spark and most warehouses have none)
CREATE UNIQUE INDEX dim_customer_one_current ON dim_customer (customer_id) WHERE is_current;

-- 1. exactly one current row per customer: catches none as well as two
SELECT customer_id, COUNT(*) FILTER (WHERE is_current) AS current_rows
FROM   dim_customer
GROUP  BY customer_id
HAVING COUNT(*) FILTER (WHERE is_current) <> 1;

-- 2. the current row is open; a closed row ends after it starts
SELECT * FROM dim_customer
WHERE  (is_current AND valid_to IS NOT NULL)
   OR  (NOT is_current AND (valid_to IS NULL OR valid_to <= valid_from));

-- 3. no gap or overlap: each version ends exactly where the next one starts
SELECT customer_id, valid_from, valid_to, next_from
FROM (
  SELECT customer_id, valid_from, valid_to,
         LEAD(valid_from) OVER (PARTITION BY customer_id ORDER BY valid_from) AS next_from
  FROM   dim_customer
) t
WHERE  next_from IS NOT NULL AND valid_to IS DISTINCT FROM next_from;

-- every query returns zero rows on a healthy load
-- 1. one row per key
SELECT order_id, COUNT(*) FROM dim_orders GROUP BY order_id HAVING COUNT(*) > 1;

-- 2. every source key arrived
SELECT DISTINCT s.order_id FROM src_orders s
WHERE  NOT EXISTS (SELECT 1 FROM dim_orders d WHERE d.order_id = s.order_id);

-- 3. totals agree with the source's latest versions
WITH latest AS (
  SELECT DISTINCT ON (order_id) order_id, amount
  FROM   src_orders ORDER BY order_id, updated_at DESC, change_seq DESC
)
SELECT (SELECT SUM(amount) FROM latest) AS source_total,
       (SELECT SUM(amount) FROM dim_orders) AS target_total
WHERE  (SELECT SUM(amount) FROM latest) IS DISTINCT FROM (SELECT SUM(amount) FROM dim_orders);