4. SCD - 고객 주소가 바뀌면 과거 주문은 어디로 배송된 걸로 남는가

-- Type 1: 현재 값으로 덮어쓴다
UPDATE dim_customers
SET city = '부산'
WHERE customer_id = 1;

-- 초기 상태: customer_id = 1, 서울
-- dim_customers_sk | customer_id | city | valid_from | valid_to   | is_current
-- 1001             | 1           | 서울 | 2025-01-01 | 9999-12-31 | true

-- 1단계: 기존 행을 닫는다
UPDATE dim_customers
SET valid_to = '2026-02-15',
    is_current = false
WHERE customer_id = 1
  AND is_current = true;

-- 2단계: 새 행을 추가한다
INSERT INTO dim_customers (dim_customers_sk, customer_id, city, valid_from, valid_to, is_current)
VALUES (1002, 1, '부산', '2026-02-15', '9999-12-31', true);

-- dim_customers_sk | customer_id | city | valid_from | valid_to   | is_current
-- 1001             | 1           | 서울 | 2025-01-01 | 2026-02-15 | false
-- 1002             | 1           | 부산 | 2026-02-15 | 9999-12-31 | true

-- 주문 시점의 고객 주소를 조인
SELECT
    o.order_id,
    o.order_date,
    c.city AS city_at_order_time
FROM fct_orders o
JOIN dim_customers c
  ON o.customer_id = c.customer_id
 AND o.order_date BETWEEN c.valid_from AND c.valid_to;

-- Type 3: 이전 값을 컬럼으로
-- customer_id | city     | previous_city
-- 1           | 부산     | 서울

UPDATE dim_customers
SET previous_city = city,
    city = '부산'
WHERE customer_id = 1;

import duckdb

conn = duckdb.connect('warehouse.duckdb')

# SCD 시연용 고객 데이터: city, membership_grade, updated_at 추가
conn.execute("""
CREATE SCHEMA IF NOT EXISTS bronze;

CREATE OR REPLACE TABLE bronze.customers_v2 AS
SELECT
    id AS customer_id,
    first_name,
    last_name,
    CASE
        WHEN id % 3 = 0 THEN '서울'
        WHEN id % 3 = 1 THEN '부산'
        ELSE '대전'
    END AS city,
    CASE
        WHEN id % 4 = 0 THEN 'Gold'
        WHEN id % 4 = 1 THEN 'Silver'
        WHEN id % 4 = 2 THEN 'Bronze'
        ELSE 'Standard'
    END AS membership_grade,
    TIMESTAMP '2025-01-15 09:00:00' AS updated_at
FROM read_csv_auto(
    'https://raw.githubusercontent.com/dbt-labs/jaffle_shop/main/seeds/raw_customers.csv'
);
""")

conn.execute("SELECT * FROM bronze.customers_v2 LIMIT 5").fetchdf()

# 변경분 데이터: 일부 고객이 이사했다
conn.execute("""
CREATE OR REPLACE TABLE bronze.customers_v2_updated AS
SELECT
    customer_id,
    first_name,
    last_name,
    CASE
        WHEN customer_id IN (1, 3, 5) THEN '제주'
        ELSE city
    END AS city,
    CASE
        WHEN customer_id IN (2, 4) THEN 'Gold'
        ELSE membership_grade
    END AS membership_grade,
    CASE
        WHEN customer_id IN (1, 2, 3, 4, 5) THEN TIMESTAMP '2026-02-20 14:00:00'
        ELSE updated_at
    END AS updated_at
FROM bronze.customers_v2;
""")

# 변경된 고객 확인
conn.execute("""
SELECT a.customer_id, a.city AS before_city, b.city AS after_city,
       a.membership_grade AS before_grade, b.membership_grade AS after_grade
FROM bronze.customers_v2 a
JOIN bronze.customers_v2_updated b ON a.customer_id = b.customer_id
WHERE a.city != b.city OR a.membership_grade != b.membership_grade
""").fetchdf()

import os
os.makedirs('jaffle_shop/snapshots', exist_ok=True)

%%writefile jaffle_shop/snapshots/snap_customers.sql
{% snapshot snap_customers %}

{{
    config(
        target_schema='snapshots',
        unique_key='customer_id',
        strategy='timestamp',
        updated_at='updated_at'
    )
}}

select * from bronze.customers_v2

{% endsnapshot %}

from dbt.cli.main import dbtRunner

# Colab의 ! 쉘 명령은 별도 프로세스를 띄운다.
# DuckDB는 프로세스 간 동시 쓰기를 막는 파일 락을 건다.
# dbtRunner로 같은 프로세스 안에서 실행하면 락 충돌이 없다.
result = dbtRunner().invoke(['snapshot', '--project-dir', 'jaffle_shop', '--profiles-dir', 'jaffle_shop'])

conn.execute("SELECT * FROM snapshots.snap_customers LIMIT 5").fetchdf()

# 소스 테이블을 변경분으로 교체
conn.execute("""
CREATE OR REPLACE TABLE bronze.customers_v2 AS
SELECT * FROM bronze.customers_v2_updated;
""")

conn.close()

# snapshot 재실행
result = dbtRunner().invoke(['snapshot', '--project-dir', 'jaffle_shop', '--profiles-dir', 'jaffle_shop'])

conn = duckdb.connect('warehouse.duckdb')

# 이력이 생성된 고객 확인
conn.execute("""
SELECT customer_id, city, membership_grade,
       dbt_valid_from, dbt_valid_to
FROM snapshots.snap_customers
WHERE customer_id IN (1, 2, 3)
ORDER BY customer_id, dbt_valid_from
""").fetchdf()

%%writefile jaffle_shop/snapshots/snap_customers_check.sql
{% snapshot snap_customers_check %}

{{
    config(
        target_schema='snapshots',
        unique_key='customer_id',
        strategy='check',
        check_cols=['city', 'membership_grade']
    )
}}

select
    customer_id,
    first_name,
    last_name,
    city,
    membership_grade
from bronze.customers_v2

{% endsnapshot %}

소스 → [Bronze] → [Silver] → [Gold]
                      ↑
         Bronze → [Snapshot] ──┘

%%writefile jaffle_shop/models/staging/stg_customers_hist.sql
with source as (
    select * from {{ ref('snap_customers') }}
),

cleaned as (
    select
        customer_id,
        first_name,
        last_name,
        city,
        membership_grade,
        dbt_valid_from AS valid_from,
        coalesce(dbt_valid_to, '9999-12-31'::timestamp) AS valid_to,
        dbt_valid_to IS NULL AS is_current
    from source
)

select * from cleaned

-- Gold: 주문 시점의 고객 정보로 조인
select
    o.order_id,
    o.order_date,
    c.city AS city_at_order,
    c.membership_grade AS grade_at_order
from stg_orders o
join stg_customers_hist c
  on o.customer_id = c.customer_id
 and o.order_date >= c.valid_from
 and o.order_date < c.valid_to

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id='medallion_with_snapshot',
    schedule='0 6 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    # 1. snapshot 먼저 실행 -Bronze의 변경 이력을 캡처
    run_snapshot = BashOperator(
        task_id='dbt_snapshot',
        bash_command='cd /opt/dbt/jaffle_shop && dbt snapshot',
    )

    # 2. Silver 변환 -snapshot 결과를 참조하는 모델이 있으니까
    run_staging = BashOperator(
        task_id='dbt_run_staging',
        bash_command='cd /opt/dbt/jaffle_shop && dbt run --select staging',
    )

    # 3. Gold 변환
    run_marts = BashOperator(
        task_id='dbt_run_marts',
        bash_command='cd /opt/dbt/jaffle_shop && dbt run --select marts',
    )

    run_snapshot >> run_staging >> run_marts