Advanced Querying with Structured Query Language Questions

WITH RECURSIVE topo_roots AS (
  -- define top-level products (those that never appear as child)
  SELECT DISTINCT parent_id AS product_id
  FROM components
  WHERE parent_id IS NOT NULL
  EXCEPT
  SELECT child_id FROM components WHERE child_id IS NOT NULL
),
bom_paths AS (
  -- seed with direct children of each top product
  SELECT
    t.product_id,
    c.child_id,
    c.qty::numeric AS qty_accum,
    ARRAY[t.product_id, c.child_id] AS path,
    FALSE AS cycle
  FROM topo_roots t
  JOIN components c ON c.parent_id = t.product_id

  UNION ALL

  -- recurse: expand children, multiply quantities, track path to detect cycles
  SELECT
    bp.product_id,
    c.child_id,
    bp.qty_accum * c.qty,
    bp.path || c.child_id,
    c.child_id = ANY(bp.path) -- mark cycle if child already in path
  FROM bom_paths bp
  JOIN components c ON c.parent_id = bp.child_id
  WHERE NOT bp.cycle -- stop expanding once a cycle detected on that branch
),
leaf_and_costs AS (
  -- choose components that have no further children (leaves) OR include intermediate if cost exists
  SELECT
    bp.product_id,
    bp.child_id AS component_id,
    bp.qty_accum,
    bp.cycle
  FROM bom_paths bp
  LEFT JOIN components cc ON cc.parent_id = bp.child_id
  WHERE cc.parent_id IS NULL -- leaf nodes
  UNION
  -- also include non-leaf components that have explicit costs
  SELECT
    bp.product_id,
    bp.child_id AS component_id,
    bp.qty_accum,
    bp.cycle
  FROM bom_paths bp
  JOIN component_costs ccost ON ccost.component_id = bp.child_id
)
SELECT
  p.product_id,
  SUM(l.qty_accum * coalesce(cc.unit_cost, 0)) AS total_cost,
  MAX(CASE WHEN l.cycle THEN 1 ELSE 0 END) AS has_cycle
FROM leaf_and_costs l
LEFT JOIN component_costs cc ON cc.component_id = l.component_id
JOIN topo_roots p ON p.product_id = l.product_id
GROUP BY p.product_id;

WITH first_event AS (
  -- user's first event timestamp and cohort week (Monday-start)
  SELECT
    user_id,
    MIN(event_ts) AS first_ts,
    date_trunc('week', MIN(event_ts))::date AS cohort_week
  FROM events
  GROUP BY user_id
),
events_by_week AS (
  -- each event mapped to its event_week (Monday-start)
  SELECT
    e.user_id,
    date_trunc('week', e.event_ts)::date AS event_week
  FROM events e
  GROUP BY 1,2
),
user_weeks AS (
  -- join to get cohort and compute week offset in weeks
  SELECT
    f.user_id,
    f.cohort_week,
    e.event_week,
    EXTRACT(EPOCH FROM (e.event_week - f.cohort_week)) / (60*60*24*7) AS week_offset_float,
    (EXTRACT(EPOCH FROM (e.event_week - f.cohort_week)) / (60*60*24*7))::int AS week_offset
  FROM first_event f
  JOIN events_by_week e USING (user_id)
),
cohort_sizes AS (
  SELECT
    cohort_week,
    COUNT(*) AS cohort_size
  FROM first_event
  GROUP BY cohort_week
),
retained AS (
  SELECT
    cohort_week,
    week_offset,
    COUNT(DISTINCT user_id) AS users_retained
  FROM user_weeks
  WHERE week_offset >= 0
  GROUP BY cohort_week, week_offset
)
SELECT
  r.cohort_week,
  r.week_offset,
  r.users_retained,
  c.cohort_size,
  ROUND( (r.users_retained::numeric / c.cohort_size) * 100, 2) AS retention_rate_pct
FROM retained r
JOIN cohort_sizes c USING (cohort_week)
ORDER BY cohort_week, week_offset;

MERGE INTO customers AS tgt
USING (
  SELECT
    canonical_key,
    payload_name,
    payload_email,
    op_type,
    event_ts,
    source,
    source_key,
    source_seq,
    source_priority,
    event_id
  FROM staging_cdc
  WHERE processed = 0
) AS src
ON tgt.canonical_key = src.canonical_key
WHEN MATCHED AND
  -- apply only if incoming event is newer or same-ts but higher priority/seq
  (
    src.event_ts > tgt.last_event_ts
    OR (src.event_ts = tgt.last_event_ts AND src.source_priority > tgt.last_source_priority)
    OR (src.event_ts = tgt.last_event_ts AND src.source_priority = tgt.last_source_priority AND src.source_seq > tgt.last_source_seq)
  )
THEN UPDATE SET
  name = CASE WHEN src.op_type = 'D' THEN NULL ELSE src.payload_name END,
  email = CASE WHEN src.op_type = 'D' THEN NULL ELSE src.payload_email END,
  deleted_flag = CASE WHEN src.op_type = 'D' THEN 1 ELSE 0 END,
  deleted_ts = CASE WHEN src.op_type = 'D' THEN src.event_ts ELSE deleted_ts END,
  last_event_ts = src.event_ts,
  last_source = src.source,
  last_source_key = src.source_key,
  last_source_seq = src.source_seq,
  last_source_priority = src.source_priority,
  last_event_id = src.event_id,
  updated_at = SYSUTCDATETIME()
WHEN NOT MATCHED BY TARGET
THEN INSERT (canonical_key, name, email, deleted_flag, deleted_ts, last_event_ts, last_source, last_source_key, last_source_seq, last_source_priority, last_event_id, created_at, updated_at)
VALUES (
  src.canonical_key,
  CASE WHEN src.op_type = 'D' THEN NULL ELSE src.payload_name END,
  CASE WHEN src.op_type = 'D' THEN NULL ELSE src.payload_email END,
  CASE WHEN src.op_type = 'D' THEN 1 ELSE 0 END,
  CASE WHEN src.op_type = 'D' THEN src.event_ts ELSE NULL END,
  src.event_ts,
  src.source,
  src.source_key,
  src.source_seq,
  src.source_priority,
  src.event_id,
  SYSUTCDATETIME(),
  SYSUTCDATETIME()
);

CREATE TABLE applied_events (
  event_id varchar PRIMARY KEY,
  canonical_key varchar,
  source varchar,
  source_key varchar,
  event_ts datetimeoffset,
  applied_at datetimeoffset
);

WITH raw_events AS (
  -- source event stream: user_id, event_type, event_ts (timestamp)
  SELECT user_id, event_type, event_ts::timestamptz FROM analytics.events
  WHERE event_type IN ('signup','onboarding_completed') OR event_type = 'conversion'
),

-- 1) Activation: find first signup and the first onboarding after signup (can be in different sessions)
first_signup AS (
  SELECT user_id, MIN(event_ts) AS signup_ts
  FROM raw_events
  WHERE event_type = 'signup'
  GROUP BY user_id
),
first_onboard_after_signup AS (
  SELECT s.user_id, MIN(e.event_ts) AS onboarding_ts
  FROM first_signup s
  JOIN raw_events e
    ON s.user_id = e.user_id
   AND e.event_type = 'onboarding_completed'
   AND e.event_ts >= s.signup_ts
  GROUP BY s.user_id
),
activation AS (
  -- activation_date is the date of onboarding completion (could choose signup date instead)
  SELECT s.user_id,
         DATE_TRUNC('day', o.onboarding_ts) AS activation_date,
         s.signup_ts, o.onboarding_ts
  FROM first_signup s
  JOIN first_onboard_after_signup o USING (user_id)
),

-- 2) Event aggregation: capture conversion events per user per day relative to activation
user_conversions AS (
  SELECT a.user_id,
         a.activation_date,
         DATE_TRUNC('day', e.event_ts) AS event_date,
         MIN(e.event_ts) AS first_conversion_ts -- if multiple conversions same day
  FROM activation a
  JOIN raw_events e
    ON a.user_id = e.user_id
   AND e.event_type = 'conversion'
   AND e.event_ts >= a.onboarding_ts -- only post-activation conversions
  GROUP BY 1,2,3
),

-- 3) Cohort metrics: compute whether user converted within N days
cohort_user_windows AS (
  SELECT a.user_id,
         a.activation_date,
         (EXTRACT(epoch FROM (uc.first_conversion_ts - a.onboarding_ts))/86400)::int AS days_to_conversion
  FROM activation a
  LEFT JOIN (
    SELECT user_id, MIN(first_conversion_ts) AS first_conversion_ts
    FROM user_conversions
    GROUP BY user_id
  ) uc USING (user_id)
),

final AS (
  SELECT
    activation_date,
    DATE_PART('week', activation_date)::int AS cohort_week,
    COUNT(*) AS cohort_size,
    COUNT(CASE WHEN days_to_conversion = 0 THEN 1 END) AS conv_day_0,
    COUNT(CASE WHEN days_to_conversion BETWEEN 0 AND 1 THEN 1 END) AS conv_1d,
    COUNT(CASE WHEN days_to_conversion BETWEEN 0 AND 6 THEN 1 END) AS conv_7d,
    COUNT(CASE WHEN days_to_conversion BETWEEN 0 AND 29 THEN 1 END) AS conv_30d
  FROM cohort_user_windows
  GROUP BY 1
  ORDER BY activation_date
)

SELECT *, conv_1d::float / cohort_size AS conv_1d_rate,
             conv_7d::float / cohort_size AS conv_7d_rate,
             conv_30d::float / cohort_size AS conv_30d_rate
FROM final;