성능 튜닝과 베스트 프랙티스

-- EXPLAIN: 실행 계획만 확인 (실제 실행 X)
EXPLAIN
SELECT * FROM orders WHERE customer_id = 1001 AND status = 'pending';

-- EXPLAIN ANALYZE: 실제 실행 후 실측값 포함
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT o.id, c.name, o.total
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.created_at > '2025-01-01'
ORDER BY o.created_at DESC
LIMIT 100;

-- 범위 파티셔닝 (Range Partitioning): 날짜 기반
CREATE TABLE events (
    id          BIGSERIAL,
    user_id     INT NOT NULL,
    action      TEXT,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- 월별 파티션 생성
CREATE TABLE events_2025_01
    PARTITION OF events
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE events_2025_02
    PARTITION OF events
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

-- 파티션 키에 자동으로 인덱스가 적용됨
-- 특정 월 조회 시 해당 파티션만 스캔 (Partition Pruning)

# Delta 테이블 → Lakebase 대량 이관 시
# Spark에서 JDBC bulk write 활용
df = spark.table("catalog.schema.large_table")

df.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/dbname") \
    .option("dbtable", "target_table") \
    .option("user", "user") \
    .option("password", "password") \
    .option("batchsize", 10000) \
    .option("numPartitions", 8) \
    .mode("append") \
    .save()

# 1. 항상 컨텍스트 매니저(with)로 트랜잭션 관리
with conn:  # 예외 발생 시 자동 rollback, 정상 시 commit
    with conn.cursor() as cur:
        cur.execute("INSERT INTO orders (...) VALUES (...)", data)

# 2. 파라미터 바인딩 — f-string 직접 삽입 금지
# ❌ SQL Injection 취약
cur.execute(f"SELECT * FROM users WHERE name = '{name}'")

# ✅ 파라미터 바인딩
cur.execute("SELECT * FROM users WHERE name = %s", (name,))

# 3. 쿼리 캐싱: 자주 바뀌지 않는 데이터는 앱 레벨에서 캐싱
@st.cache_data(ttl=300)  # 5분 캐시
def get_product_categories():
    with engine.connect() as conn:
        return pd.read_sql("SELECT DISTINCT category FROM products ORDER BY 1", conn)

# 4. 커서 기반 페이지네이션
@st.cache_data(ttl=30)
def get_recent_orders(last_id=None, page_size=50):
    query = "SELECT * FROM orders"
    params = []
    if last_id:
        query += " WHERE id < %s"
        params.append(last_id)
    query += " ORDER BY id DESC LIMIT %s"
    params.append(page_size)
    with engine.connect() as conn:
        return pd.read_sql(query, conn, params=params)

# customer_app.py — 고객 관리 CRUD 앱 (최적화 적용)
import streamlit as st
import psycopg2
import psycopg2.extras
import psycopg2.pool
import pandas as pd

st.set_page_config(page_title="고객 관리", page_icon="👥", layout="wide")

@st.cache_resource
def get_pool():
    """앱 수명 동안 커넥션 풀을 한 번만 생성합니다."""
    return psycopg2.pool.ThreadedConnectionPool(
        minconn=2, maxconn=10,
        host=st.secrets["lakebase"]["host"],
        port=5432,
        dbname=st.secrets["lakebase"]["dbname"],
        user=st.secrets["lakebase"]["user"],
        password=st.secrets["lakebase"]["password"],
        sslmode="require"
    )

def get_conn():
    return get_pool().getconn()

def release_conn(conn):
    get_pool().putconn(conn)

@st.cache_data(ttl=300)
def get_tier_list():
    """등급 목록 — 5분 캐시."""
    conn = get_conn()
    try:
        return pd.read_sql("SELECT DISTINCT tier FROM customers ORDER BY 1", conn)["tier"].tolist()
    finally:
        release_conn(conn)

st.title("👥 고객 관리 시스템")
tab_list, tab_create, tab_edit = st.tabs(["📋 고객 목록", "➕ 신규 등록", "✏️ 정보 수정"])

# --- 조회 (Read) ---
with tab_list:
    search = st.text_input("🔍 고객 검색 (이름 또는 이메일)")
    conn = get_conn()
    try:
        if search:
            df = pd.read_sql(
                "SELECT id, name, email, tier, created_at FROM customers "
                "WHERE name ILIKE %s OR email ILIKE %s ORDER BY created_at DESC",
                conn, params=(f"%{search}%", f"%{search}%")
            )
        else:
            df = pd.read_sql(
                "SELECT id, name, email, tier, created_at FROM customers "
                "ORDER BY created_at DESC LIMIT 100",
                conn
            )
    finally:
        release_conn(conn)
    st.dataframe(df, use_container_width=True)
    st.caption(f"총 {len(df)}명 표시 중")

# --- 생성 (Create) ---
with tab_create:
    with st.form("create_customer"):
        name  = st.text_input("이름")
        email = st.text_input("이메일")
        tier  = st.selectbox("등급", get_tier_list())
        if st.form_submit_button("등록", type="primary"):
            conn = get_conn()
            try:
                with conn:
                    with conn.cursor() as cur:
                        cur.execute(
                            "INSERT INTO customers (name, email, tier) VALUES (%s, %s, %s)",
                            (name, email, tier)
                        )
                st.success(f"✅ {name}님이 등록되었습니다!")
            except psycopg2.errors.UniqueViolation:
                st.error("❌ 이미 등록된 이메일입니다.")
            finally:
                release_conn(conn)

# --- 수정 / 삭제 (Update / Delete) ---
with tab_edit:
    customer_id = st.number_input("고객 ID", min_value=1, step=1)
    conn = get_conn()
    try:
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(
                "SELECT id, name, email, tier FROM customers WHERE id = %s",
                (customer_id,)
            )
            customer = cur.fetchone()
    finally:
        release_conn(conn)

    if customer:
        with st.form("edit_customer"):
            new_name = st.text_input("이름", value=customer["name"])
            tier_opts = get_tier_list()
            new_tier = st.selectbox(
                "등급", tier_opts,
                index=tier_opts.index(customer["tier"]) if customer["tier"] in tier_opts else 0
            )
            col1, col2 = st.columns(2)
            with col1:
                if st.form_submit_button("수정 저장", type="primary"):
                    conn = get_conn()
                    try:
                        with conn:
                            with conn.cursor() as cur:
                                cur.execute(
                                    "UPDATE customers SET name=%s, tier=%s, updated_at=NOW() WHERE id=%s",
                                    (new_name, new_tier, customer_id)
                                )
                        st.success("✅ 수정되었습니다!")
                    finally:
                        release_conn(conn)
            with col2:
                if st.form_submit_button("삭제 (소프트)", type="secondary"):
                    conn = get_conn()
                    try:
                        with conn:
                            with conn.cursor() as cur:
                                cur.execute(
                                    "UPDATE customers SET is_deleted=TRUE, deleted_at=NOW() WHERE id=%s",
                                    (customer_id,)
                                )
                        st.warning("🗑️ 삭제 처리되었습니다.")
                    finally:
                        release_conn(conn)
    else:
        st.info("조회된 고객이 없습니다. ID를 확인하십시오.")