Complex Joins and Set Operations Questions

WITH
-- prepare small dims once; force broadcast by keeping them small or using OPTIONS in BigQuery
dim_customer AS (
  SELECT customer_id, country, cohort, vip_flag
  FROM analytics.dim_customer
  WHERE active = TRUE
),
dim_product AS (
  SELECT product_id, category, price_tier
  FROM analytics.dim_product
),
-- narrow fact scan by applying filters and projecting only needed columns early
filtered_facts AS (
  SELECT order_id, customer_id, product_id, order_ts, qty, revenue
  FROM analytics.fact_orders
  WHERE order_ts BETWEEN '2025-01-01' AND '2025-01-31'
    AND region = 'US'             -- cluster/partitioned column for pruning
)
SELECT f.order_id, f.order_ts, c.country, c.vip_flag, p.category, f.qty, f.revenue
FROM filtered_facts f
JOIN dim_customer c USING (customer_id)   -- broadcast join on small dim
JOIN dim_product p USING (product_id);

-- 1) Pre-aggregate sessions per user-week
CREATE OR REPLACE TABLE analytics.sessions_weekly
PARTITION BY week_start
CLUSTER BY user_id AS
SELECT
  DATE_TRUNC(session_start, WEEK) AS week_start,
  user_id,
  COUNT(*) AS sessions,
  MIN(session_start) AS first_session_ts
FROM raw.sessions
WHERE session_start >= DATE_SUB(CURRENT_DATE(), INTERVAL 365 DAY)
GROUP BY week_start, user_id;

SELECT
  u.acq_week,
  COUNT(DISTINCT u.user_id) AS cohort_users,
  COUNT(DISTINCT s.user_id) AS users_with_session,
  COUNT(DISTINCT c.user_id) AS users_with_click,
  COUNT(DISTINCT p.user_id) AS users_with_purchase,
  SAFE_DIVIDE(COUNT(DISTINCT s.user_id), COUNT(DISTINCT u.user_id)) AS pct_session,
  SAFE_DIVIDE(COUNT(DISTINCT p.user_id), COUNT(DISTINCT u.user_id)) AS pct_purchase
FROM users_by_week u -- small dim: user_id, acq_week (partitioned by acq_week)
LEFT JOIN analytics.sessions_weekly s
  ON s.user_id = u.user_id AND s.week_start = u.acq_week
LEFT JOIN analytics.clicks_weekly c
  ON c.user_id = u.user_id AND c.week_start = u.acq_week
LEFT JOIN analytics.purchases_weekly p
  ON p.user_id = u.user_id AND p.week_start = u.acq_week
GROUP BY u.acq_week;

-- ensure index for fast lookup
CREATE INDEX IF NOT EXISTS idx_profiles_user_updated ON user_profiles(user_id, updated_at DESC);

SELECT e.*,
       p.profile_id,
       p.updated_at,
       p.profile_json
FROM user_events e
LEFT JOIN LATERAL (
  SELECT profile_id, updated_at, profile_json
  FROM user_profiles up
  WHERE up.user_id = e.user_id
    AND up.updated_at <= e.occurred_at
  ORDER BY up.updated_at DESC
  LIMIT 1
) p ON true;

WITH events_with_profiles AS (
  SELECT e.*, up.profile_id, up.updated_at,
         ROW_NUMBER() OVER (PARTITION BY e.event_id ORDER BY up.updated_at DESC) rn
  FROM user_events e
  LEFT JOIN user_profiles up
    ON up.user_id = e.user_id AND up.updated_at <= e.occurred_at
)
SELECT * EXCEPT (rn) FROM events_with_profiles WHERE rn = 1;

-- 1. prepare staging (already loaded): stg_customers(customer_id, name, email, ... , change_hash, src_load_date)

-- 2. candidate deltas: new inserts and changed rows
WITH current AS (
  SELECT id as dim_id, customer_id, change_hash
  FROM dim_customers
  WHERE current_flag = TRUE
),
delta AS (
  SELECT s.*
  FROM stg_customers s
  LEFT JOIN current c ON s.customer_id = c.customer_id
  WHERE c.customer_id IS NULL OR s.change_hash <> c.change_hash
)
-- 3a. For databases supporting MERGE (Snowflake/Postgres 15+)
MERGE INTO dim_customers d
USING delta s
  ON d.customer_id = s.customer_id AND d.current_flag = TRUE
WHEN MATCHED THEN
  -- close existing current row
  UPDATE SET current_flag = FALSE, end_date = s.src_load_date - INTERVAL '1 SECOND'
WHEN NOT MATCHED THEN
  -- insert new history row (will also run for matched after update)
  INSERT (customer_id, name, email, change_hash, start_date, end_date, current_flag, created_at)
  VALUES (s.customer_id, s.name, s.email, s.change_hash, s.src_load_date, NULL, TRUE, CURRENT_TIMESTAMP);