CDC 처리 — APPLY CHANGES

APPLY CHANGES INTO target_table
FROM source
KEYS (key_columns)           -- 레코드를 식별하는 기본 키
SEQUENCE BY sequence_column  -- 변경 순서를 결정하는 컬럼
[COLUMNS ...]                -- 포함/제외할 컬럼
[WHERE ...]                  -- 필터 조건
[APPLY AS DELETE WHEN ...]   -- 삭제 조건
[APPLY AS TRUNCATE WHEN ...] -- 전체 삭제 조건
STORED AS SCD TYPE {1|2}     -- SCD 유형
[TRACK HISTORY ON ...]       -- (Type 2) 이력 추적 대상 컬럼

-- 1단계: 타겟 Streaming Table 생성
CREATE OR REFRESH STREAMING TABLE silver_customers;

-- 2단계: CDC 적용 (SCD Type 1)
APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data, _metadata)
STORED AS SCD TYPE 1;

import dlt

dlt.create_streaming_table("silver_customers")

@dlt.apply_changes(
    target="silver_customers",
    source="bronze_customer_cdc",
    keys=["customer_id"],
    sequence_by=col("updated_at"),
    except_column_list=["_rescued_data", "_metadata"],
    stored_as_scd_type=1
)
def apply_customer_changes():
    pass

-- 1단계: 타겟 Streaming Table 생성
CREATE OR REFRESH STREAMING TABLE silver_customers_history;

-- 2단계: CDC 적용 (SCD Type 2)
APPLY CHANGES INTO silver_customers_history
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data, _metadata)
STORED AS SCD TYPE 2;

-- 현재 유효한 레코드만 필터링
SELECT *
FROM silver_customers_history
WHERE __END_AT IS NULL;

-- 특정 시점의 고객 정보 조회 (Time Travel 스타일)
SELECT *
FROM silver_customers_history
WHERE __START_AT <= '2024-02-01'
  AND (__END_AT > '2024-02-01' OR __END_AT IS NULL);

APPLY CHANGES INTO silver_customers_history
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data)
STORED AS SCD TYPE 2
TRACK HISTORY ON (city, email);  -- city나 email이 변경될 때만 이력 생성

APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data)
APPLY AS DELETE WHEN operation = 'DELETE'
STORED AS SCD TYPE 2;

APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS * EXCEPT (_rescued_data)
APPLY AS DELETE WHEN operation = 'DELETE'
STORED AS SCD TYPE 1;

-- Debezium CDC 이벤트 처리
APPLY CHANGES INTO silver_orders
FROM STREAM(bronze_debezium_orders)
KEYS (payload.after.order_id)
SEQUENCE BY payload.source.ts_ms
COLUMNS
    payload.after.order_id AS order_id,
    payload.after.customer_id AS customer_id,
    payload.after.amount AS amount,
    payload.after.status AS status,
    payload.after.updated_at AS updated_at
APPLY AS DELETE WHEN payload.op = 'd'
STORED AS SCD TYPE 1;

-- Lakeflow Connect가 수집한 CDC 데이터 처리
APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_mysql_customers)
KEYS (customer_id)
SEQUENCE BY _metadata.ingestion_timestamp
COLUMNS * EXCEPT (_metadata, _rescued_data)
APPLY AS DELETE WHEN _change_type = 'delete'
STORED AS SCD TYPE 2;

-- 커스텀 변경 로그 테이블 처리
APPLY CHANGES INTO silver_products
FROM STREAM(bronze_product_changelog)
KEYS (product_id)
SEQUENCE BY change_sequence_no
COLUMNS product_id, name, price, category, status
APPLY AS DELETE WHEN change_type = 'D'
WHERE change_type IN ('I', 'U', 'D')  -- INSERT, UPDATE, DELETE만 처리
STORED AS SCD TYPE 1;

-- ============================================
-- Bronze: 원본 CDC 이벤트 수집 (Auto Loader)
-- ============================================
CREATE OR REFRESH STREAMING TABLE bronze_customer_cdc
AS SELECT
    *,
    _metadata.file_path AS source_file,
    _metadata.file_modification_time AS file_time
FROM STREAM read_files(
    '/volumes/catalog/schema/landing/customers/',
    format => 'json',
    schema => '
        customer_id BIGINT,
        name STRING,
        email STRING,
        city STRING,
        phone STRING,
        operation STRING,
        updated_at TIMESTAMP
    '
);

-- ============================================
-- Silver (SCD Type 1): 고객 현재 상태
-- ============================================
CREATE OR REFRESH STREAMING TABLE silver_customers;

APPLY CHANGES INTO silver_customers
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS customer_id, name, email, city, phone
APPLY AS DELETE WHEN operation = 'DELETE'
STORED AS SCD TYPE 1;

-- ============================================
-- Silver (SCD Type 2): 고객 이력
-- ============================================
CREATE OR REFRESH STREAMING TABLE silver_customers_history;

APPLY CHANGES INTO silver_customers_history
FROM STREAM(bronze_customer_cdc)
KEYS (customer_id)
SEQUENCE BY updated_at
COLUMNS customer_id, name, email, city, phone
APPLY AS DELETE WHEN operation = 'DELETE'
STORED AS SCD TYPE 2
TRACK HISTORY ON (city, email);  -- 주소, 이메일 변경만 추적

-- ============================================
-- Gold: 도시별 고객 분포 (현재 기준)
-- ============================================
CREATE OR REFRESH MATERIALIZED VIEW gold_customer_by_city
AS SELECT
    city,
    COUNT(*) AS customer_count,
    COUNT(DISTINCT email) AS unique_emails
FROM silver_customers
GROUP BY city
ORDER BY customer_count DESC;

-- ============================================
-- Gold: 고객 이동 이력 분석
-- ============================================
CREATE OR REFRESH MATERIALIZED VIEW gold_customer_city_changes
AS SELECT
    customer_id,
    name,
    city AS current_city,
    __START_AT AS moved_at,
    LAG(city) OVER (PARTITION BY customer_id ORDER BY __START_AT) AS previous_city
FROM silver_customers_history
WHERE __END_AT IS NULL OR __END_AT IS NOT NULL
ORDER BY customer_id, __START_AT;