SQL for Data Analysis Questions

ANALYZE orders;
ANALYZE customers;

CREATE STATISTICS st ON (col_a, col_b) FROM schema.table;
ANALYZE schema.table;

SELECT o.*
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE c.region = 'North' AND o.created_at >= '2025-01-01';

SELECT o.*
FROM (
  SELECT id FROM customers WHERE region = 'North'
) c
JOIN orders o ON o.customer_id = c.id
WHERE o.created_at >= '2025-01-01';

SELECT o.*
FROM orders o
WHERE o.created_at >= '2025-01-01'
  AND EXISTS (SELECT 1 FROM customers c WHERE c.id = o.customer_id AND c.region = 'North');

-- Parameters: choose decay rate (lambda per hour). Here lambda = 0.1
WITH all_touches AS (
  SELECT
    c.conversion_id,
    c.user_id,
    c.conversion_time,
    c.revenue,
    t.channel,
    t.touch_time,
    EXTRACT(EPOCH FROM (c.conversion_time - t.touch_time))/3600.0 AS hours_before  -- hours between touch and conversion
  FROM conversions c
  JOIN touchpoints t
    ON c.user_id = t.user_id
  WHERE t.touch_time <= c.conversion_time
),

first_touch AS (
  SELECT DISTINCT ON (conversion_id)
    conversion_id, channel, revenue
  FROM all_touches
  ORDER BY conversion_id, touch_time ASC
),

last_touch AS (
  SELECT DISTINCT ON (conversion_id)
    conversion_id, channel, revenue
  FROM all_touches
  ORDER BY conversion_id, touch_time DESC
),

time_decay_weights AS (
  SELECT
    conversion_id,
    channel,
    revenue,
    hours_before,
    EXP(-0.1 * GREATEST(hours_before, 0)) AS raw_weight  -- lambda=0.1 per hour
  FROM all_touches
),

time_decay_norm AS (
  SELECT
    td.conversion_id,
    td.channel,
    td.revenue * td.raw_weight / SUM(td.raw_weight) OVER (PARTITION BY td.conversion_id) AS attributed_revenue
  FROM time_decay_weights td
)

-- Results: attribution per conversion_id and channel for each model
SELECT 'first_touch' AS model, conversion_id, channel, SUM(revenue) AS attributed_revenue
FROM first_touch
GROUP BY conversion_id, channel

UNION ALL

SELECT 'last_touch' AS model, conversion_id, channel, SUM(revenue) AS attributed_revenue
FROM last_touch
GROUP BY conversion_id, channel

UNION ALL

SELECT 'time_decay' AS model, conversion_id, channel, SUM(attributed_revenue) AS attributed_revenue
FROM time_decay_norm
GROUP BY conversion_id, channel
ORDER BY model, conversion_id;

-- events: id, event_ts_utc (timestamptz)
-- profiles: user_id, user_timezone (text, IANA name like 'America/Los_Angeles')

SELECT
  e.id,
  e.user_id,
  e.event_ts_utc,
  -- convert UTC timestamptz to local timestamp without time zone
  (e.event_ts_utc AT TIME ZONE p.user_timezone) AS local_ts,
  -- compute business_date: if local time < 04:00, subtract 1 day, else take date
  (CASE
     WHEN (e.event_ts_utc AT TIME ZONE p.user_timezone)::time < time '04:00'
     THEN ( (e.event_ts_utc AT TIME ZONE p.user_timezone)::date - INTERVAL '1 day')::date
     ELSE (e.event_ts_utc AT TIME ZONE p.user_timezone)::date
   END) AS business_date
FROM events e
JOIN profiles p USING (user_id);

BEGIN;
-- serialize batch application to avoid concurrent double-apply
SELECT pg_advisory_xact_lock(hashtext(:batch_id));

-- abort if batch already applied
INSERT INTO processed_batches(batch_id, status, applied_at)
VALUES (:batch_id,'in_progress', now())
ON CONFLICT DO NOTHING;

WITH already AS (
  SELECT 1 FROM processed_batches WHERE batch_id = :batch_id AND status = 'done'
)
SELECT CASE WHEN exists (SELECT 1 FROM already) THEN RAISE EXCEPTION 'batch already applied' ELSE 1 END;

INSERT INTO target_table(id, payload, deleted, last_batch_id, last_seq, updated_at)
SELECT id, payload, tombstone, batch_id, seq, now()
FROM incoming_rows
ON CONFLICT (id) DO UPDATE
SET
  payload = CASE
    WHEN (EXCLUDED.last_batch_id = target_table.last_batch_id AND EXCLUDED.last_seq <= target_table.last_seq) THEN target_table.payload
    WHEN (EXCLUDED.last_batch_id IS NULL) THEN target_table.payload
    ELSE EXCLUDED.payload END,
  deleted = CASE
    WHEN (EXCLUDED.last_batch_id = target_table.last_batch_id AND EXCLUDED.last_seq <= target_table.last_seq) THEN target_table.deleted
    ELSE EXCLUDED.deleted END,
  last_batch_id = GREATEST(target_table.last_batch_id, EXCLUDED.last_batch_id),
  last_seq = CASE WHEN EXCLUDED.last_batch_id = target_table.last_batch_id THEN GREATEST(target_table.last_seq, EXCLUDED.last_seq) ELSE EXCLUDED.last_seq END,
  updated_at = now();

UPDATE processed_batches SET status='done', applied_at=now() WHERE batch_id=:batch_id;
COMMIT;