SQL for Data Analysis Questions

-- landing table receiving CDC events
CREATE TABLE cdc_events (
  src_table text,
  src_pk jsonb,         -- primary key as JSON for composite keys
  op_type char(1),      -- 'I','U','D'
  payload jsonb,        -- full row after change (for I/U) or null for D
  src_ts timestamptz,   -- source commit timestamp
  lsn bigint,           -- source log sequence number for ordering
  schema_version int
);

MERGE INTO dim_customer tgt
USING (
  SELECT
    (payload->>'customer_id')::int as customer_id,
    payload as payload,
    src_ts, lsn
  FROM cdc_events
  WHERE src_table='customer' AND op_type IN ('I','U')
) src
ON tgt.customer_id = src.customer_id AND tgt.current_flag = true
WHEN MATCHED AND (
     (tgt.name IS DISTINCT FROM (src.payload->>'name'))
  OR (tgt.email IS DISTINCT FROM (src.payload->>'email'))
) THEN
  -- expire old row
  UPDATE SET current_flag = false, end_ts = src.src_ts
WHEN NOT MATCHED THEN
  -- insert new version
  INSERT (customer_id, name, email, start_ts, end_ts, current_flag, lsn, schema_version)
  VALUES (src.customer_id, src.payload->>'name', src.payload->>'email', src.src_ts, '9999-12-31'::timestamptz, true, src.lsn, (src.payload->>'schema_version')::int);

-- dedup using (event_id) and only accept events newer than watermark
INSERT INTO fact_orders (order_id, amount, event_ts, lsn)
SELECT (payload->>'order_id')::int, (payload->>'amount')::numeric, src_ts, lsn
FROM cdc_events e
WHERE src_table='orders' AND op_type='I'
  AND NOT EXISTS (
    SELECT 1 FROM fact_orders f WHERE f.event_id = (e.payload->>'event_id')
  )
  AND e.src_ts > (SELECT watermark_ts FROM ingestion_watermark WHERE table_name='orders');

-- Range partitioning by month (pseudocode)
CREATE TABLE events (...) 
PARTITION BY RANGE (date_trunc('month', event_ts));

-- staging_recent contains new/updated/deleted events for watermark window
MERGE INTO mv_agg AS T
USING (
  SELECT user_id, DATE_TRUNC('minute', event_time) AS minute, COUNT(*) AS cnt, SUM(value) AS sum_value
  FROM staging_recent
  GROUP BY user_id, DATE_TRUNC('minute', event_time)
) AS S
ON T.user_id = S.user_id AND T.minute = S.minute
WHEN MATCHED THEN
  UPDATE SET cnt = T.cnt + S.cnt, sum_value = T.sum_value + S.sum_value, updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN
  INSERT (user_id, minute, cnt, sum_value, updated_at)
  VALUES (S.user_id, S.minute, S.cnt, S.sum_value, CURRENT_TIMESTAMP);

-- 1) Pre-aggregate heavy fact tables
CREATE TEMP TABLE user_events_agg AS
SELECT user_id,
       COUNT(*) AS events_cnt,
       SUM(value) AS events_value,
       MAX(event_ts) AS last_event_ts
FROM events
WHERE event_ts >= now() - interval '90 days'
GROUP BY user_id;

CREATE TEMP TABLE user_purchases_agg AS
SELECT user_id,
       COUNT(*) AS purchases_cnt,
       SUM(amount) AS purchases_sum
FROM purchases
WHERE status = 'complete'
GROUP BY user_id;

-- 2) Join pre-aggregates and compute score with a single window
CREATE TEMP TABLE user_scores AS
SELECT u.user_id,
       u.events_cnt,
       p.purchases_cnt,
       p.purchases_sum,
       -- complex score computed once
       (0.6 * COALESCE(p.purchases_sum,0) / NULLIF(u.events_cnt,0)
        + 0.4 * EXTRACT(EPOCH FROM (now() - u.last_event_ts))/86400) AS score
FROM users u
LEFT JOIN user_events_agg uea ON u.user_id = uea.user_id
LEFT JOIN user_purchases_agg p ON u.user_id = p.user_id;

-- 3) Top-100 using single window or ORDER/LIMIT with index
SELECT user_id, events_cnt, purchases_cnt, purchases_sum, score
FROM user_scores
ORDER BY score DESC
LIMIT 100;

WITH daily AS (
  -- 1 row per user per day (date type); make sure to create an index on (user_id, day)
  SELECT user_id, (event_ts::date) AS day
  FROM events
  GROUP BY user_id, (event_ts::date)
)
SELECT
  d.user_id,
  d.day,
  -- count of active days in [day - 29 days, day]
  (SELECT COUNT(1)
   FROM daily d2
   WHERE d2.user_id = d.user_id
     AND d2.day BETWEEN d.day - INTERVAL '29 days' AND d.day
  ) AS active_days_30d
FROM daily d
ORDER BY d.user_id, d.day;