07 — Common Table Expressions (CTEs)

-- Single CTE: name a subquery, reference it in the main query.
-- The CTE 'big_orders' is a named subquery whose result is available in the main SELECT.
-- PostgreSQL 12+ inlines non-recursive CTEs by default (folds into main query like a macro).
WITH big_orders AS (
  SELECT customer_id, SUM(amount) AS total
  FROM orders
  GROUP BY customer_id
  HAVING SUM(amount) > 100
)
SELECT c.name, b.total
FROM big_orders b
JOIN customers c ON c.id = b.customer_id
ORDER BY b.total DESC;

-- Pipeline: customer_totals → city_totals → final filter.
-- Each CTE reads the prior one. This reads like English: "compute totals, then city averages, then filter."
WITH
  customer_totals AS (
    SELECT customer_id, SUM(amount) AS total
    FROM orders
    GROUP BY customer_id
  ),
  city_totals AS (
    -- References customer_totals (defined above) — forward references are allowed.
    SELECT c.city, AVG(ct.total) AS avg_customer_spend
    FROM customer_totals ct
    JOIN customers c ON c.id = ct.customer_id
    GROUP BY c.city
  )
SELECT * FROM city_totals
WHERE avg_customer_spend > 50
ORDER BY avg_customer_spend DESC;

-- Compare each customer's spend to the overall mean and median.
-- 'totals' is referenced THREE times: main query + two scalar subqueries.
-- With MATERIALIZED (PG12+), it's computed once and reused; without, it may be inlined.
WITH totals AS (
  SELECT customer_id, SUM(amount) AS total
  FROM orders
  GROUP BY customer_id
)
SELECT
  t.customer_id,
  t.total,
  t.total - (SELECT AVG(total) FROM totals) AS diff_from_mean,
  t.total - (
    SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY total) FROM totals
  ) AS diff_from_median
FROM totals t
ORDER BY t.total DESC;

-- MATERIALIZED: compute once, store result, reuse.
-- Use when: expensive computation + referenced multiple times.
-- Warning: prevents predicate pushdown from the outer query into the CTE.
WITH expensive AS MATERIALIZED (
  SELECT id, complex_function(data) AS result FROM big_table  -- runs ONCE, full scan
)
SELECT * FROM expensive WHERE result > 0.5;

-- NOT MATERIALIZED: force inlining even when planner would materialize
-- (e.g., recursive CTEs or multiply-referenced CTEs with costly re-evaluation).
WITH cheap AS NOT MATERIALIZED (
  SELECT id, name FROM small_table
)
SELECT * FROM cheap WHERE name LIKE 'A%';

-- Archive old orders: CTE identifies rows, INSERT copies them.
WITH old_orders AS (
  SELECT id FROM orders WHERE ordered_on < '2023-01-01'
)
INSERT INTO orders_archive
SELECT * FROM orders WHERE id IN (SELECT id FROM old_orders);

-- Update the top spender's status: CTE finds the top spender, UPDATE applies.
WITH top AS (
  SELECT customer_id
  FROM orders
  GROUP BY customer_id
  ORDER BY SUM(amount) DESC
  LIMIT 1
)
UPDATE customers SET status = 'vip'
WHERE id IN (SELECT customer_id FROM top);

-- Atomic "move": DELETE from orders, INSERT into archive, all in one statement.
-- The DELETE's RETURNING rows feed the INSERT. Both run in the same transaction.
-- Note: data-modifying CTEs see the PRE-statement snapshot — they don't see
-- each other's changes within the same statement (only RETURNING rows are visible).
WITH moved AS (
  DELETE FROM orders WHERE ordered_on < '2023-01-01'
  RETURNING *
)
INSERT INTO orders_archive SELECT * FROM moved;

-- Multi-table atomic operation: insert customer, use RETURNING id for their first order.
WITH new_customer AS (
  INSERT INTO customers (name, email)
  VALUES ('Alice', 'alice@example.com')
  RETURNING id
)
INSERT INTO orders (customer_id, amount, ordered_on)
SELECT id, 100.00, CURRENT_DATE
FROM new_customer;

-- Production ETL: stage → dedup → aggregate → upsert.
-- Each CTE is a named stage in the pipeline. All run in one atomic statement.
WITH
  -- Stage 1: Read raw events from staging table, filter invalid rows.
  staged AS (
    SELECT
      event_id,
      customer_id,
      product_id,
      amount,
      event_ts,
      source                                  -- track provenance for debugging
    FROM raw_events_staging
    WHERE event_ts >= '2024-01-01'
      AND customer_id IS NOT NULL             -- drop rows with NULL FK (would poison later joins)
      AND amount > 0                          -- drop negative/zero amounts (data quality)
  ),

  -- Stage 2: Deduplicate by event_id (keep latest by event_ts).
  -- ROW_NUMBER partitioned by event_id, ordered by event_ts DESC → rn=1 is the latest.
  deduped AS (
    SELECT *
    FROM (
      SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY event_ts DESC) AS rn
      FROM staged
    ) ranked
    WHERE rn = 1                               -- only the latest version of each event
  ),

  -- Stage 3: Aggregate to daily customer-product totals.
  aggregated AS (
    SELECT
      customer_id,
      product_id,
      DATE(event_ts) AS event_date,
      SUM(amount) AS daily_total,
      COUNT(*) AS event_count
    FROM deduped
    GROUP BY customer_id, product_id, DATE(event_ts)
  ),

  -- Stage 4: Upsert into the fact table (PostgreSQL ON CONFLICT).
  -- INSERT new rows; on conflict, UPDATE the daily_total and event_count.
  upserted AS (
    INSERT INTO fact_daily_sales (customer_id, product_id, event_date, daily_total, event_count)
    SELECT customer_id, product_id, event_date, daily_total, event_count
    FROM aggregated
    ON CONFLICT (customer_id, product_id, event_date)
    DO UPDATE SET
      daily_total = EXCLUDED.daily_total,
      event_count = EXCLUDED.event_count,
      updated_at = CURRENT_TIMESTAMP
    RETURNING id, customer_id, product_id, event_date
  )

-- Final: log what was upserted (for audit/monitoring).
SELECT COUNT(*) AS rows_upserted, COUNT(DISTINCT customer_id) AS customers_affected
FROM upserted;

-- Three levels of nesting: unreadable, unmaintainable, can't inspect intermediate stages.
SELECT city, AVG(customer_total) AS avg_spend
FROM (
  SELECT c.city, c.id, SUM(o.amount) AS customer_total
  FROM (
    SELECT * FROM customers WHERE status = 'active'
  ) c
  JOIN orders o ON c.id = o.customer_id
  GROUP BY c.city, c.id
) per_customer
GROUP BY city
HAVING AVG(customer_total) > (
  SELECT AVG(total) FROM (
    SELECT SUM(amount) AS total FROM orders GROUP BY customer_id
  ) avg_per_customer
);

-- Each stage is named and inspectable. Comment out the final SELECT and
-- replace with 'SELECT * FROM aggregated' to debug any intermediate stage.
WITH active_customers AS (
  SELECT * FROM customers WHERE status = 'active'
),
customer_totals AS (
  SELECT c.city, c.id, SUM(o.amount) AS customer_total
  FROM active_customers c
  JOIN orders o ON c.id = o.customer_id
  GROUP BY c.city, c.id
),
overall_avg AS (
  SELECT AVG(customer_total) AS avg_total FROM customer_totals
)
SELECT c.city, AVG(c.customer_total) AS avg_spend
FROM customer_totals c
CROSS JOIN overall_avg oa
GROUP BY c.city
HAVING AVG(c.customer_total) > oa.avg_total
ORDER BY avg_spend DESC;

WITH
  inserted AS (
    INSERT INTO customers (name, email)
    VALUES ('Bob', 'bob@example.com')
    RETURNING id, name
  ),
  audit_log AS (
    -- Intent: log that a new customer was added by reading the customers table.
    SELECT id, name FROM customers WHERE name = 'Bob'
  )
SELECT * FROM audit_log;