CTEs & Subqueries Questions

-- Assume preagg.product_sales_monthly(partition_date, product_id, qty, revenue, updated_at)
WITH
-- A: recent source rows since last_processed_ts (parameter)
recent_facts AS (
  SELECT product_id,
         DATE_TRUNC('month', sale_ts) AS partition_date,
         SUM(quantity) AS qty,
         SUM(quantity * price) AS revenue,
         MAX(updated_at) AS max_updated_at
  FROM fact_sales
  WHERE sale_ts >= @last_processed_ts
  GROUP BY 1,2
),
-- B: existing aggregate rows for affected partitions/keys
existing_agg AS (
  SELECT product_id, partition_date, qty, revenue
  FROM preagg.product_sales_monthly
  WHERE partition_date IN (SELECT DISTINCT partition_date FROM recent_facts)
),
-- C: compute merged values (delta upsert logic)
deltas AS (
  SELECT r.product_id,
         r.partition_date,
         COALESCE(e.qty,0) + r.qty AS new_qty,
         COALESCE(e.revenue,0) + r.revenue AS new_revenue,
         r.max_updated_at
  FROM recent_facts r
  LEFT JOIN existing_agg e
    ON r.product_id = e.product_id
   AND r.partition_date = e.partition_date
)
-- Final: atomic merge into partitioned pre-aggregate table
MERGE INTO preagg.product_sales_monthly AS tgt
USING deltas AS src
  ON tgt.product_id = src.product_id
 AND tgt.partition_date = src.partition_date
WHEN MATCHED THEN
  UPDATE SET qty = src.new_qty, revenue = src.new_revenue, updated_at = src.max_updated_at
WHEN NOT MATCHED THEN
  INSERT (partition_date, product_id, qty, revenue, updated_at)
  VALUES (src.partition_date, src.product_id, src.new_qty, src.new_revenue, src.max_updated_at);

-- 1) Normalize & mark source metadata, 2) Deduplicate staging keeping latest event per business key,
-- 3) MERGE into target table (events) updating if newer or inserting if new.

WITH staged AS (
  SELECT
    se.event_id,
    se.user_id,
    se.event_type,
    se.event_timestamp,        -- actual event time (may be late)
    se.arrival_ts,             -- when row landed in staging
    se.payload,
    CURRENT_TIMESTAMP() AS processed_ts
  FROM staging_events se
  WHERE se.event_timestamp IS NOT NULL
),

deduped AS (
  -- Keep the single best row per event_id (or business key)
  SELECT DISTINCT ON (event_id) *
  FROM staged
  ORDER BY event_id, event_timestamp DESC, arrival_ts DESC
),

upsert_source AS (
  -- enrich deduped rows with target lookup to decide action
  SELECT
    d.*,
    t.event_timestamp AS target_event_ts,
    t.payload AS target_payload
  FROM deduped d
  LEFT JOIN events t
    ON t.event_id = d.event_id
)

-- Snowflake/Postgres (Postgres >=15 supports MERGE). For Postgres <15, use INSERT ... ON CONFLICT.
MERGE INTO events AS tgt
USING upsert_source AS src
  ON tgt.event_id = src.event_id
WHEN MATCHED AND (src.event_timestamp > tgt.event_timestamp OR tgt.event_timestamp IS NULL) THEN
  UPDATE SET
    user_id = src.user_id,
    event_type = src.event_type,
    event_timestamp = src.event_timestamp,
    payload = src.payload,
    updated_at = src.processed_ts
WHEN NOT MATCHED THEN
  INSERT (event_id, user_id, event_type, event_timestamp, payload, created_at)
  VALUES (src.event_id, src.user_id, src.event_type, src.event_timestamp, src.payload, src.processed_ts);

WITH users AS (
  -- base table: user_id, signup_at, event_at (activity timestamp)
  SELECT user_id,
         DATE_TRUNC('week', signup_at)::date AS signup_week,
         DATE_TRUNC('week', event_at)::date AS event_week
  FROM user_events
  WHERE event_at IS NOT NULL
),
cohort_weeks AS (
  -- compute week offset per activity
  SELECT
    user_id,
    signup_week,
    event_week,
    DATE_PART('week', event_week::timestamp)  -- not used directly; use weeks_between
      AS dummy,
    FLOOR(EXTRACT(EPOCH FROM (event_week - signup_week)) / (7*24*3600))::int AS week_offset
  FROM users
),
filtered AS (
  -- only week 0..12 and dedupe per user-week (active if any activity)
  SELECT DISTINCT user_id, signup_week, week_offset
  FROM cohort_weeks
  WHERE week_offset BETWEEN 0 AND 12
),
cohort_sizes AS (
  -- total users in each cohort (signup week)
  SELECT signup_week, COUNT(DISTINCT user_id) AS cohort_size
  FROM users
  GROUP BY signup_week
),
retention AS (
  -- active users per cohort-week
  SELECT
    f.signup_week,
    f.week_offset,
    COUNT(DISTINCT f.user_id) AS active_users
  FROM filtered f
  GROUP BY f.signup_week, f.week_offset
),
retention_pivot AS (
  -- pivot weeks into columns 0..12
  SELECT
    r.signup_week,
    c.cohort_size,
    MAX(CASE WHEN week_offset = 0 THEN active_users END) AS w0,
    MAX(CASE WHEN week_offset = 1 THEN active_users END) AS w1,
    MAX(CASE WHEN week_offset = 2 THEN active_users END) AS w2,
    MAX(CASE WHEN week_offset = 3 THEN active_users END) AS w3,
    MAX(CASE WHEN week_offset = 4 THEN active_users END) AS w4,
    MAX(CASE WHEN week_offset = 5 THEN active_users END) AS w5,
    MAX(CASE WHEN week_offset = 6 THEN active_users END) AS w6,
    MAX(CASE WHEN week_offset = 7 THEN active_users END) AS w7,
    MAX(CASE WHEN week_offset = 8 THEN active_users END) AS w8,
    MAX(CASE WHEN week_offset = 9 THEN active_users END) AS w9,
    MAX(CASE WHEN week_offset = 10 THEN active_users END) AS w10,
    MAX(CASE WHEN week_offset = 11 THEN active_users END) AS w11,
    MAX(CASE WHEN week_offset = 12 THEN active_users END) AS w12
  FROM retention r
  JOIN cohort_sizes c USING (signup_week)
  GROUP BY r.signup_week, c.cohort_size
  ORDER BY signup_week DESC
)
SELECT
  signup_week,
  cohort_size,
  w0, w1, w2, w3, w4, w5, w6, w7, w8, w9, w10, w11, w12,
  -- retention rates as percentages (optional)
  ROUND(100.0 * COALESCE(w0,0) / cohort_size, 2) AS pct_w0,
  ROUND(100.0 * COALESCE(w1,0) / cohort_size, 2) AS pct_w1,
  ROUND(100.0 * COALESCE(w12,0) / cohort_size, 2) AS pct_w12
FROM retention_pivot;

CREATE INDEX ON orders (order_date DESC);
CREATE INDEX ON orders (customer_id);

-- Instead of WITH cte_customers AS (...) SELECT ... FROM orders JOIN cte_customers USING (customer_id)
-- Inline
SELECT ... FROM orders o JOIN (
  SELECT customer_id, segment FROM customers WHERE active
) c ON o.customer_id = c.customer_id
WHERE o.order_date >= '2025-01-01';

WITH cte_customers AS MATERIALIZED (...) -- or NOT MATERIALIZED

CREATE TABLE orders_y2025_q1 PARTITION OF orders FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

CREATE INDEX ON orders (order_date) INCLUDE (total_amount, status);