Trino SQL 실전 패턴 — Window, UNNEST, Lambda, 그리고 UDF

-- 카테고리별 매출 상위 3개 상품
SELECT category, product, revenue
FROM (
  SELECT category, product, revenue,
         row_number() OVER (PARTITION BY category ORDER BY revenue DESC) AS rn
  FROM sales
)
WHERE rn <= 3;

SELECT
  d,
  revenue,
  sum(revenue) OVER (ORDER BY d ROWS UNBOUNDED PRECEDING) AS running_total,
  avg(revenue) OVER (ORDER BY d ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS ma7
FROM daily_sales;

-- 직전 이벤트와의 시간 간격
SELECT
  user_id, event_time,
  event_time - lag(event_time) OVER (PARTITION BY user_id ORDER BY event_time) AS gap
FROM events;

WITH marked AS (
  SELECT user_id, event_time,
         CASE WHEN event_time - lag(event_time)
                   OVER (PARTITION BY user_id ORDER BY event_time)
                   > INTERVAL '30' MINUTE
              THEN 1 ELSE 0 END AS is_new_session
  FROM events
)
SELECT user_id, event_time,
       sum(is_new_session) OVER (PARTITION BY user_id ORDER BY event_time) AS session_id
FROM marked;

-- 한 주문에 여러 상품(배열)이 든 행을 상품 단위 행으로
SELECT order_id, item
FROM orders
CROSS JOIN UNNEST(items) AS t(item);
 
-- 인덱스(순번)와 함께 펼치기 (WITH ORDINALITY)
SELECT order_id, idx, item
FROM orders
CROSS JOIN UNNEST(items) WITH ORDINALITY AS t(item, idx);
 
-- MAP 펼치기 (키, 값)
SELECT id, k, v
FROM tbl
CROSS JOIN UNNEST(attributes) AS t(k, v);

SELECT order_id, array_agg(item ORDER BY item) AS items
FROM order_items
GROUP BY order_id;

SELECT
  json_extract_scalar(payload, '$.user.id')      AS user_id,
  json_extract_scalar(payload, '$.event.type')   AS event_type,
  cast(json_extract(payload, '$.items') AS ARRAY(VARCHAR)) AS items
FROM raw_events;

SELECT
  transform(prices, x -> x * 1.1)                 AS with_tax,   -- map
  filter(scores, x -> x >= 60)                    AS passed,     -- filter
  reduce(nums, 0, (s, x) -> s + x, s -> s)        AS total,      -- fold/sum
  array_sort(tags)                                AS sorted_tags,
  any_match(flags, x -> x = true)                 AS has_true,
  cardinality(filter(scores, x -> x >= 60))       AS pass_count
FROM t;

-- 오브젝트 스토리지의 이벤트(Iceberg) × RDBMS 의 고객 마스터(PostgreSQL)
SELECT e.event_type, c.tier, count(*) AS cnt
FROM iceberg.analytics.events e
JOIN postgresql.crm.customers c ON e.user_id = c.id
WHERE e.event_time >= TIMESTAMP '2026-06-01 00:00:00 UTC'
GROUP BY e.event_type, c.tier;
 
-- 한 시스템의 데이터를 다른 시스템으로 적재 (CTAS 페더레이션)
CREATE TABLE iceberg.analytics.customer_snapshot AS
SELECT * FROM postgresql.crm.customers;

-- 안전한 형변환 (실패 시 NULL)
try_cast(value AS INTEGER)
 
-- NULL 안전 비교 / 치환
coalesce(a, b, 0)
nullif(a, 0)             -- a=0 이면 NULL (0으로 나누기 방지)
 
-- 날짜 (단위가 첫 인자!)
date_add('day', 7, current_date)
date_diff('hour', t1, t2)
date_trunc('month', event_time)
format_datetime(event_time, 'yyyy-MM-dd')
 
-- 문자열
split(path, '/')                      -- 배열로
regexp_extract(ua, '(\d+)\.(\d+)', 1)
url_extract_host(url)
 
-- 근사 집계 (대용량에서 빠름)
approx_distinct(user_id)              -- 근사 distinct count
approx_percentile(latency, 0.95)      -- p95

-- 쿼리 안에서 즉석 정의 (WITH FUNCTION)
WITH FUNCTION to_won(usd DOUBLE)
  RETURNS DOUBLE
  RETURN usd * 1330.0
SELECT product, to_won(price_usd) AS price_krw
FROM catalog.schema.products;

CREATE FUNCTION catalog.schema.mask_email(email VARCHAR)
RETURNS VARCHAR
RETURN regexp_replace(email, '(^.).*(@.*$)', '$1***$2');

CREATE FUNCTION catalog.schema.py_slug(s VARCHAR)
RETURNS VARCHAR
LANGUAGE PYTHON
WITH (handler = 'slug')
AS $$
def slug(s):
    return "-".join(s.lower().split()) if s else None
$$;