Complex Data Integration and Joins Questions

-- enable trigram
CREATE EXTENSION IF NOT EXISTS pg_trgm;

-- example normalization function (could be in ETL instead)
CREATE OR REPLACE FUNCTION normalize_phone(text) RETURNS text AS $$
  SELECT regexp_replace($1, '\D', '', 'g');
$$ LANGUAGE sql IMMUTABLE;

-- indexes
CREATE INDEX ON customers_a (normalize_phone(raw_phone));
CREATE INDEX ON customers_b (normalize_phone(raw_phone));
CREATE INDEX ON customers_a USING gin (name gin_trgm_ops);
CREATE INDEX ON customers_b USING gin (name gin_trgm_ops);

-- fuzzy join with blocking and similarity threshold
SELECT a.*, b.*, similarity(a.name, b.name) AS name_sim
FROM customers_a a
JOIN customers_b b
  ON normalize_phone(a.raw_phone) = normalize_phone(b.raw_phone) -- primary block
  AND left(lower(a.name),1) = left(lower(b.name),1)            -- cheap additional block
  AND abs(length(a.name) - length(b.name)) <= 4                -- length filter
WHERE similarity(a.name, b.name) > 0.4                          -- trigram similarity threshold
ORDER BY a.id, name_sim DESC;

SELECT a.*, b.*
FROM customers_a a
CROSS JOIN LATERAL (
  SELECT b.*, similarity(a.name,b.name) AS sim
  FROM customers_b b
  WHERE normalize_phone(b.raw_phone)=normalize_phone(a.raw_phone)
    AND similarity(a.name,b.name) > 0.4
  ORDER BY sim DESC
  LIMIT 5
) b;

WITH user_event AS (
  SELECT v.user_id, e.event_type
  FROM visits v
  JOIN events e USING (visit_id)
  GROUP BY v.user_id, e.event_type  -- one row per user x event_type
),
user_conv AS (
  SELECT user_id, 1 AS converted
  FROM conversions
  GROUP BY user_id
)
SELECT
  ue.event_type,
  COUNT(*) AS users_with_event,
  COALESCE(SUM(uc.converted),0) AS users_converted,
  COALESCE(SUM(uc.converted)::float / COUNT(*), 0) AS conversion_rate
FROM user_event ue
LEFT JOIN user_conv uc USING (user_id)
GROUP BY ue.event_type
ORDER BY conversion_rate DESC;

WITH user_event AS (
  SELECT user_id, event_type
  FROM (
    SELECT v.user_id, e.event_type,
           ROW_NUMBER() OVER (PARTITION BY v.user_id, e.event_type ORDER BY v.visit_id) rn
    FROM visits v JOIN events e USING (visit_id)
  ) WHERE rn = 1
),
user_conv AS (
  SELECT user_id, 1 AS converted
  FROM conversions
  QUALIFY ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY conversion_id) = 1
)
-- same aggregation as above

SELECT s.cleaned_row_id, s.normalized_email, array_agg(c.customer_id) as candidates
FROM cleaned_staging s
JOIN customers c ON c.normalized_email = s.normalized_email
GROUP BY s.cleaned_row_id, s.normalized_email
HAVING count(*) > 1;

SELECT e.employee_id,
       e.name AS employee_name,
       m.employee_id  AS manager_id,
       m.name         AS manager_name
FROM employees e
LEFT JOIN employees m
  ON e.manager_id = m.employee_id
ORDER BY e.employee_id;

WITH RECURSIVE chain(start_id, employee_id, path, depth) AS (
  SELECT e.employee_id, e.employee_id, ARRAY[e.employee_id], 0
  FROM employees e
  UNION ALL
  SELECT c.start_id, m.employee_id, path || m.employee_id, depth + 1
  FROM chain c
  JOIN employees m ON m.employee_id = (SELECT manager_id FROM employees WHERE employee_id = c.employee_id)
  WHERE m.employee_id IS NOT NULL
    AND NOT m.employee_id = ANY(path)    -- stop if cycle would repeat
    AND depth < 1000                     -- safeguard for very deep chains
)
SELECT start_id AS employee_in_cycle
FROM chain
WHERE employee_id = ANY(path)             -- indicates a repetition
GROUP BY start_id;