Join Operations and Multi Table Queries Questions

WITH
-- 1) cohort: user's signup week
cohort AS (
  SELECT user_id,
         DATE_TRUNC('week', signup_date)::date AS cohort_week
  FROM users
),

-- 2) pre-agg events per user-week (reduces row volume)
events_week AS (
  SELECT user_id,
         DATE_TRUNC('week', event_time)::date AS week,
         COUNT(*) AS events_count,
         MAX(event_time) AS last_event_at
  FROM events
  GROUP BY 1,2
),

-- 3) purchases per user-week
purchases_week AS (
  SELECT user_id,
         DATE_TRUNC('week', purchase_time)::date AS week,
         COUNT(*) AS purchases_count,
         SUM(amount) AS purchases_amount
  FROM purchases
  GROUP BY 1,2
),

-- 4) churn signals per user-week (flag if churned that week)
churn_week AS (
  SELECT user_id,
         DATE_TRUNC('week', churn_date)::date AS week,
         1 AS churned_flag
  FROM churns
  WHERE churn_date IS NOT NULL
  GROUP BY 1,2
),

-- 5) calendar of weeks to expand cohort x offsets (0..3)
weeks AS (
  SELECT DISTINCT DATE_TRUNC('week', signup_date)::date AS cohort_week FROM users
),
expanded AS (
  SELECT w.cohort_week,
         (w.cohort_week + (interval '1 week' * offset))::date AS week,
         offset
  FROM weeks w
  CROSS JOIN (VALUES (0),(1),(2),(3)) AS t(offset)
),

-- 6) join cohort users to expanded weeks and left-join pre-aggregates
cohort_activity AS (
  SELECT c.cohort_week,
         e.offset AS week_offset,
         c.user_id,
         CASE WHEN ew.user_id IS NOT NULL THEN 1 ELSE 0 END AS active_flag,
         COALESCE(pw.purchases_count,0) AS purchases_count,
         COALESCE(ch.churned_flag,0) AS churned_flag
  FROM cohort c
  JOIN expanded e ON e.cohort_week = c.cohort_week
  LEFT JOIN events_week ew
    ON ew.user_id = c.user_id AND ew.week = e.week
  LEFT JOIN purchases_week pw
    ON pw.user_id = c.user_id AND pw.week = e.week
  LEFT JOIN churn_week ch
    ON ch.user_id = c.user_id AND ch.week = e.week
),

-- 7) aggregate per cohort_week & week_offset and compute retention
retention AS (
  SELECT
    cohort_week,
    week_offset,
    COUNT(DISTINCT user_id) FILTER (WHERE week_offset = 0) OVER (PARTITION BY cohort_week) AS cohort_size,
    SUM(active_flag) AS active_users
  FROM cohort_activity
  GROUP BY cohort_week, week_offset
)

SELECT
  cohort_week,
  week_offset,
  active_users,
  ROUND(100.0 * active_users::numeric / NULLIF(cohort_size,0),2) AS retention_rate_pct
FROM retention
ORDER BY cohort_week, week_offset;

WITH src1 AS (
  SELECT id, LOWER(TRIM(name)) AS name_norm, REGEXP_REPLACE(phone,'\D','','g') AS phone_norm,
         SUBSTR(REGEXP_REPLACE(LOWER(TRIM(name)),'[^a-z]',''),1,1) AS first_initial,
         RIGHT(REGEXP_REPLACE(LOWER(TRIM(name)),'[^a-z]',''),4) AS last4
  FROM customers_a
),
src2 AS (
  SELECT id, LOWER(TRIM(name)) AS name_norm, REGEXP_REPLACE(phone,'\D','','g') AS phone_norm,
         SUBSTR(REGEXP_REPLACE(LOWER(TRIM(name)),'[^a-z]',''),1,1) AS first_initial,
         RIGHT(REGEXP_REPLACE(LOWER(TRIM(name)),'[^a-z]',''),4) AS last4
  FROM customers_b
),
candidates AS (
  SELECT a.id AS a_id, b.id AS b_id, a.name_norm AS a_name, b.name_norm AS b_name,
         a.phone_norm, b.phone_norm
  FROM src1 a
  JOIN src2 b
    ON a.first_initial = b.first_initial
   AND a.last4 = b.last4
   AND COALESCE(a.phone_norm, '') = COALESCE(b.phone_norm, '') -- optional
)
SELECT c.*, levenshtein(a_name, b_name) AS name_dist,
       1.0 - (CAST(levenshtein(a_name,b_name) AS FLOAT) / GREATEST(LENGTH(a_name),LENGTH(b_name))) AS name_sim
FROM candidates c
WHERE levenshtein(a_name, b_name) <= 2  -- threshold tuned to balance precision/recall
   OR (1.0 - (levenshtein(a_name,b_name)::float / GREATEST(LENGTH(a_name),LENGTH(b_name)))) > 0.85;

-- pre-aggregate order_items to order totals, then join to customers
WITH order_totals AS (
  SELECT order_id, SUM(quantity * unit_price) AS order_revenue
  FROM order_items
  GROUP BY order_id
)
SELECT c.customer_id, c.name, SUM(ot.order_revenue) AS total_revenue
FROM customers c
JOIN orders o ON o.customer_id = c.customer_id
JOIN order_totals ot ON ot.order_id = o.order_id
GROUP BY c.customer_id, c.name;

-- remove duplicate order_item rows (keep first occurrence) then sum
SELECT customer_id, SUM(order_item_amount) AS total_revenue
FROM (
  SELECT c.customer_id,
         oi.order_item_id,
         oi.quantity * oi.unit_price AS order_item_amount,
         ROW_NUMBER() OVER (PARTITION BY oi.order_item_id ORDER BY oi.updated_at DESC) AS rn
  FROM customers c
  JOIN orders o    ON o.customer_id = c.customer_id
  JOIN order_items oi ON oi.order_id = o.order_id
  -- maybe joined to another child table that caused duplicates
) t
WHERE rn = 1
GROUP BY customer_id;

SELECT country, COUNT(*) AS rows
FROM events
GROUP BY country
ORDER BY rows DESC
LIMIT 10;

ALTER TABLE events ADD COLUMN salt INT;
UPDATE events
SET salt = CASE WHEN country='US' THEN CAST(rand()*8 AS INT) ELSE 0 END;

SELECT *
FROM events e
JOIN customers c
  ON e.country = c.country AND e.salt = COALESCE(c.salt,0);

/* Example in Spark SQL */
SELECT /*+ BROADCAST(c) */ *
FROM events e
JOIN customers c ON e.country = c.country;

-- Spark example
events.repartitionByRange(col("country"), col("user_id"))

-- pseudo:
WITH us AS (SELECT * FROM events WHERE country='US'),
     rest AS (SELECT * FROM events WHERE country<>'US')
-- process us with added partitioning/salt, then UNION results with rest