-- 부서별 급여 순위와 누적 합계
SELECT
    employee_id,
    name,
    department,
    salary,
    RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
    SUM(salary) OVER (PARTITION BY department) AS dept_total,
    AVG(salary) OVER (
        PARTITION BY department
        ORDER BY hire_date
        ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) AS rolling_avg_3
FROM employees;

-- 월별 매출 증감률 계산
SELECT
    month,
    revenue,
    LAG(revenue, 1) OVER (ORDER BY month) AS prev_month,
    ROUND(
        (revenue - LAG(revenue, 1) OVER (ORDER BY month)) * 100.0
        / NULLIF(LAG(revenue, 1) OVER (ORDER BY month), 0),
        2
    ) AS growth_rate_pct
FROM monthly_sales;

-- 재귀 CTE로 조직도 탐색
WITH RECURSIVE org_tree AS (
    -- Base case: 최상위 직원
    SELECT
        employee_id,
        name,
        manager_id,
        0 AS depth,
        name::TEXT AS path
    FROM employees
    WHERE manager_id IS NULL

    UNION ALL

    -- Recursive case
    SELECT
        e.employee_id,
        e.name,
        e.manager_id,
        ot.depth + 1,
        ot.path || ' > ' || e.name
    FROM employees e
    INNER JOIN org_tree ot ON e.manager_id = ot.employee_id
)
SELECT employee_id, name, depth, path
FROM org_tree
ORDER BY path;

-- 복잡한 분석을 단계별 CTE로 분리
WITH
top_customers AS (
    SELECT customer_id, SUM(amount) AS total_spent
    FROM orders
    WHERE created_at >= NOW() - INTERVAL '90 days'
    GROUP BY customer_id
    HAVING SUM(amount) > 1000
),
customer_details AS (
    SELECT c.*, tc.total_spent
    FROM customers c
    JOIN top_customers tc ON c.id = tc.customer_id
),
ranked AS (
    SELECT *,
           NTILE(4) OVER (ORDER BY total_spent DESC) AS quartile
    FROM customer_details
)
SELECT * FROM ranked WHERE quartile = 1;

EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT u.name, COUNT(o.id) AS order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.created_at > '2025-01-01'
GROUP BY u.id, u.name
ORDER BY order_count DESC
LIMIT 10;

Limit  (cost=1234.56..1234.57 rows=10 width=72) (actual time=45.123..45.125 rows=10 loops=1)
  ->  Sort  (cost=1234.56..1259.56 rows=10000 width=72) (actual time=45.120..45.121 rows=10 loops=1)
        Sort Key: (count(o.id)) DESC
        Sort Method: top-N heapsort  Memory: 25kB
        ->  HashAggregate  (cost=876.00..976.00 rows=10000 width=72) (actual time=38.456..42.234 rows=8523 loops=1)
              Group Key: u.id
              Batches: 1  Memory Usage: 1553kB
              ->  Hash Left Join  (cost=345.00..801.00 rows=15000 width=40) (actual time=5.678..28.901 rows=15000 loops=1)
                    Hash Cond: (o.user_id = u.id)
                    ->  Seq Scan on orders o  (cost=0.00..312.00 rows=15000 width=16) (actual time=0.023..8.456 rows=15000 loops=1)
                    ->  Hash  (cost=270.00..270.00 rows=6000 width=32) (actual time=5.234..5.234 rows=6000 loops=1)
                          Buckets: 8192  Batches: 1  Memory Usage: 358kB
                          ->  Index Scan using idx_users_created_at on users u  (cost=0.29..270.00 rows=6000 width=32)
Planning Time: 1.234 ms
Execution Time: 45.456 ms

-- 복합 인덱스: 선택도 높은 컬럼을 앞에
CREATE INDEX idx_orders_user_status_date
ON orders (user_id, status, created_at DESC);

-- 부분 인덱스: 특정 조건만 인덱싱 (공간 절약)
CREATE INDEX idx_active_users
ON users (email)
WHERE deleted_at IS NULL AND status = 'active';

-- 표현식 인덱스: 함수 결과 인덱싱
CREATE INDEX idx_users_lower_email
ON users (LOWER(email));

-- BRIN 인덱스: 시계열 데이터에 효율적 (크기 매우 작음)
CREATE INDEX idx_logs_timestamp_brin
ON application_logs USING BRIN (created_at);

-- GIN 인덱스: 배열, JSONB, 전문검색에 적합
CREATE INDEX idx_products_tags_gin
ON products USING GIN (tags);

-- 트랜잭션 격리 수준 설정
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;

BEGIN;
  -- 계좌 이체: 원자성 보장
  UPDATE accounts SET balance = balance - 500 WHERE id = 1;
  UPDATE accounts SET balance = balance + 500 WHERE id = 2;

  -- 잔액 검증
  DO $$
  DECLARE
    bal NUMERIC;
  BEGIN
    SELECT balance INTO bal FROM accounts WHERE id = 1;
    IF bal < 0 THEN
      RAISE EXCEPTION 'Insufficient funds';
    END IF;
  END $$;

COMMIT;

-- JSONB 컬럼 생성과 GIN 인덱스
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    metadata JSONB NOT NULL DEFAULT '{}'
);

CREATE INDEX idx_products_metadata ON products USING GIN (metadata);

-- JSONB 연산자
-- ->> : 텍스트로 추출
-- -> : JSON으로 추출
-- @> : containment 체크
-- ? : 키 존재 여부

SELECT * FROM products
WHERE metadata @> '{"category": "electronics", "in_stock": true}';

SELECT
    name,
    metadata->>'brand' AS brand,
    (metadata->>'price')::NUMERIC AS price,
    metadata->'specs'->>'cpu' AS cpu
FROM products
WHERE metadata ? 'discount_pct'
  AND (metadata->>'discount_pct')::NUMERIC > 10;

-- JSONB 업데이트: 특정 키만 변경
UPDATE products
SET metadata = jsonb_set(
    metadata,
    '{price}',
    '29900'::jsonb
)
WHERE id = 42;

-- JSONB 경로 쿼리
SELECT jsonb_path_query(
    metadata,
    '$.specs.memory ? (@ > 16)'
) FROM products;

-- 시계열 데이터 레인지 파티셔닝
CREATE TABLE events (
    id BIGSERIAL,
    user_id INT,
    event_type TEXT,
    payload JSONB,
    created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);

-- 월별 파티션 자동 생성 함수
CREATE OR REPLACE FUNCTION create_monthly_partition(target_date DATE)
RETURNS VOID AS $$
DECLARE
    partition_name TEXT;
    start_date DATE;
    end_date DATE;
BEGIN
    start_date := DATE_TRUNC('month', target_date);
    end_date := start_date + INTERVAL '1 month';
    partition_name := 'events_' || TO_CHAR(start_date, 'YYYY_MM');

    EXECUTE FORMAT(
        'CREATE TABLE IF NOT EXISTS %I PARTITION OF events
         FOR VALUES FROM (%L) TO (%L)',
        partition_name, start_date, end_date
    );
END;
$$ LANGUAGE plpgsql;

SELECT create_monthly_partition('2026-03-01');
SELECT create_monthly_partition('2026-04-01');

-- 파티션 프루닝 확인
EXPLAIN SELECT * FROM events
WHERE created_at BETWEEN '2026-03-01' AND '2026-03-31';

-- Publisher 설정
ALTER SYSTEM SET wal_level = logical;

CREATE PUBLICATION app_publication
FOR TABLE users, orders, products
WITH (publish = 'insert, update, delete');

-- Subscriber에서 구독 설정
CREATE SUBSCRIPTION app_subscription
CONNECTION 'host=primary-db port=5432 dbname=myapp user=replicator'
PUBLICATION app_publication;

-- 복제 상태 모니터링
SELECT
    subname,
    received_lsn,
    latest_end_lsn,
    latest_end_time
FROM pg_stat_subscription;

import redis
import json
import hashlib
from functools import wraps
from typing import Any, Optional

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Cache-Aside 패턴 (Lazy Loading)
def get_user_profile(user_id: int) -> dict:
    cache_key = f"user:profile:{user_id}"

    # 1. 캐시 확인
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. DB에서 조회
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)

    # 3. 캐시에 저장 (TTL 1시간)
    r.setex(cache_key, 3600, json.dumps(user))
    return user

# Write-Through 패턴: 쓰기 시 캐시도 동시 업데이트
def update_user_profile(user_id: int, data: dict) -> None:
    cache_key = f"user:profile:{user_id}"

    # DB 업데이트
    db.execute("UPDATE users SET ... WHERE id = %s", user_id)

    # 캐시도 즉시 업데이트
    updated = get_user_from_db(user_id)
    r.setex(cache_key, 3600, json.dumps(updated))

# Write-Behind (Write-Back) 패턴: 비동기 DB 저장
class WriteBehindCache:
    def __init__(self):
        self.dirty_keys_set = "cache:dirty_keys"

    def write(self, key: str, value: dict, ttl: int = 3600):
        # 캐시에 즉시 쓰기
        r.setex(key, ttl, json.dumps(value))
        # 더티 키 목록에 추가 (나중에 DB에 flush)
        r.sadd(self.dirty_keys_set, key)

    def flush_to_db(self):
        dirty_keys = r.smembers(self.dirty_keys_set)
        for key in dirty_keys:
            data = r.get(key)
            if data:
                db.upsert(json.loads(data))
                r.srem(self.dirty_keys_set, key)

# 분산 락 (Redlock)
import time

def acquire_lock(lock_name: str, timeout: int = 10) -> Optional[str]:
    identifier = str(time.time())
    lock_key = f"lock:{lock_name}"
    acquired = r.set(lock_key, identifier, nx=True, ex=timeout)
    return identifier if acquired else None

def release_lock(lock_name: str, identifier: str) -> bool:
    lock_key = f"lock:{lock_name}"
    lua_script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(lua_script, 1, lock_key, identifier)
    return bool(result)

# Sorted Set: 실시간 랭킹 보드
def update_score(player: str, score: int):
    r.zadd("leaderboard", {player: score})

def get_top_players(n: int = 10):
    return r.zrevrange("leaderboard", 0, n-1, withscores=True)

# HyperLogLog: 유니크 방문자 수 (근사값, 메모리 효율적)
def track_visitor(page: str, user_id: str):
    r.pfadd(f"visitors:{page}:{today()}", user_id)

def get_unique_visitors(page: str, date: str) -> int:
    return r.pfcount(f"visitors:{page}:{date}")

# Pub/Sub: 실시간 알림
import threading

def publisher():
    for i in range(10):
        r.publish("notifications", json.dumps({"type": "alert", "msg": f"Event {i}"}))

def subscriber():
    pubsub = r.pubsub()
    pubsub.subscribe("notifications")
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            print(f"Received: {data}")

from pymongo import MongoClient, ASCENDING, DESCENDING
from datetime import datetime

client = MongoClient('mongodb://localhost:27017/')
db = client['ecommerce']

# 임베디드 도큐먼트 vs 레퍼런스 설계
# 임베디드: 자주 함께 조회되는 데이터
orders_collection = db['orders']
sample_order = {
    "_id": "order_12345",
    "user_id": "user_67890",
    "status": "shipped",
    "created_at": datetime.utcnow(),
    # 주소는 임베디드 (변경 이력 보존)
    "shipping_address": {
        "street": "서울시 강남구 테헤란로 123",
        "city": "서울",
        "zip": "06234"
    },
    # 상품 스냅샷 임베디드 (가격 변동 보존)
    "items": [
        {"product_id": "p001", "name": "노트북", "price": 1200000, "qty": 1},
        {"product_id": "p002", "name": "마우스", "price": 35000, "qty": 2}
    ],
    "total": 1270000
}

# Aggregation Pipeline
pipeline = [
    # 1단계: 최근 30일 주문 필터
    {"$match": {
        "created_at": {"$gte": datetime(2026, 2, 17)},
        "status": {"$in": ["delivered", "shipped"]}
    }},
    # 2단계: 아이템 언와인드
    {"$unwind": "$items"},
    # 3단계: 상품별 집계
    {"$group": {
        "_id": "$items.product_id",
        "product_name": {"$first": "$items.name"},
        "total_qty": {"$sum": "$items.qty"},
        "total_revenue": {"$sum": {"$multiply": ["$items.price", "$items.qty"]}}
    }},
    # 4단계: 매출 기준 정렬
    {"$sort": {"total_revenue": -1}},
    # 5단계: 상위 10개
    {"$limit": 10},
    # 6단계: 결과 재구성
    {"$project": {
        "product_id": "$_id",
        "product_name": 1,
        "total_qty": 1,
        "total_revenue": 1,
        "_id": 0
    }}
]

top_products = list(orders_collection.aggregate(pipeline))

-- Cassandra CQL: 쿼리 패턴 중심 테이블 설계
-- "쿼리를 먼저 정의하고, 테이블을 거기에 맞춘다"

-- 사용자별 타임라인 조회 쿼리를 위한 테이블
CREATE TABLE user_timeline (
    user_id UUID,
    created_at TIMEUUID,
    post_id UUID,
    content TEXT,
    likes COUNTER,
    PRIMARY KEY (user_id, created_at)
) WITH CLUSTERING ORDER BY (created_at DESC)
  AND compaction = {
    'class': 'TimeWindowCompactionStrategy',
    'compaction_window_unit': 'DAYS',
    'compaction_window_size': 7
  };

-- 태그별 게시물 검색 테이블 (별도로 비정규화)
CREATE TABLE posts_by_tag (
    tag TEXT,
    created_at TIMEUUID,
    post_id UUID,
    user_id UUID,
    title TEXT,
    PRIMARY KEY (tag, created_at, post_id)
) WITH CLUSTERING ORDER BY (created_at DESC);

-- pgvector 확장 설치
CREATE EXTENSION IF NOT EXISTS vector;

-- 임베딩 저장 테이블
CREATE TABLE document_embeddings (
    id BIGSERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}',
    embedding vector(1536),  -- OpenAI text-embedding-3-small 차원
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW 인덱스 (고속 ANN 검색)
CREATE INDEX idx_doc_embeddings_hnsw
ON document_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- 코사인 유사도 검색
SELECT
    id,
    content,
    metadata,
    1 - (embedding <=> '[0.1, 0.2, ...]'::vector) AS similarity
FROM document_embeddings
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 5;

-- 메타데이터 필터링과 벡터 검색 결합 (Hybrid Search)
SELECT
    id,
    content,
    metadata->>'source' AS source,
    1 - (embedding <=> query_embedding) AS similarity
FROM document_embeddings
WHERE
    metadata->>'language' = 'ko'
    AND metadata->>'category' = 'technical'
    AND (embedding <=> query_embedding) < 0.3  -- 임계값 필터
ORDER BY embedding <=> query_embedding
LIMIT 10;

-- IVFFlat: 빌드가 빠르고 메모리 효율적
-- 대용량 데이터셋 초기 구축에 적합
CREATE INDEX idx_ivfflat
ON document_embeddings
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);  -- sqrt(행 수) 권장

-- 검색 품질 조정
SET ivfflat.probes = 10;  -- 검색 클러스터 수 (높을수록 정확, 느림)

-- HNSW: 검색 속도가 빠르고 recall 높음
-- 빌드 시간이 길고 메모리 많이 사용
CREATE INDEX idx_hnsw
ON document_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

SET hnsw.ef_search = 40;  -- 검색 시 탐색 폭 (높을수록 정확)

강한 일관성 (Strong Consistency)
   ↓ 성능 저하, 지연 증가
순차 일관성 (Sequential Consistency)
   ↓
인과 일관성 (Causal Consistency)
   ↓
최종 일관성 (Eventual Consistency)
   ↓ 성능 향상, 가용성 증가
단조 읽기 (Monotonic Read)

# Range Sharding: 연속 범위로 분할
def range_shard(user_id: int, num_shards: int = 4) -> int:
    shard_size = 250_000_000  # 10억 사용자를 4개 샤드
    return min(user_id // shard_size, num_shards - 1)

# Hash Sharding: 균등 분배 (핫스팟 방지)
import hashlib

def hash_shard(key: str, num_shards: int = 8) -> int:
    hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
    return hash_val % num_shards

# Consistent Hashing: 노드 추가/제거 시 리밸런싱 최소화
import bisect

class ConsistentHashRing:
    def __init__(self, nodes: list, replicas: int = 150):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            self.add_node(node)

    def add_node(self, node: str):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            bisect.insort(self.sorted_keys, key)

    def get_node(self, key: str) -> str:
        if not self.ring:
            return None
        hash_val = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, hash_val)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.ring[self.sorted_keys[idx]]

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

-- 정규화된 스키마
CREATE TABLE categories (id SERIAL PRIMARY KEY, name TEXT);
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT,
    category_id INT REFERENCES categories(id)
);

-- 비정규화된 스키마 (읽기 최적화)
CREATE TABLE products_denormalized (
    id SERIAL PRIMARY KEY,
    name TEXT,
    category_id INT,
    category_name TEXT  -- 중복 저장으로 JOIN 제거
);

-- 팩트 테이블 (측정값)
CREATE TABLE fact_sales (
    sale_id BIGINT,
    date_key INT,
    product_key INT,
    customer_key INT,
    store_key INT,
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2)
);

-- 디멘션 테이블 (맥락 정보)
CREATE TABLE dim_date (
    date_key INT PRIMARY KEY,
    full_date DATE,
    year INT, quarter INT, month INT, week INT, day_of_week INT
);

CREATE TABLE dim_product (
    product_key INT PRIMARY KEY,
    product_id TEXT,
    name TEXT, category TEXT, brand TEXT, unit_cost DECIMAL(10,2)
);

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_postgres import PGVector
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.schema import Document

# 1. 임베딩 모델 설정
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 2. pgvector 벡터 스토어 연결
CONNECTION_STRING = "postgresql+psycopg://user:password@localhost:5432/vectordb"
COLLECTION_NAME = "documents"

vector_store = PGVector(
    embeddings=embeddings,
    collection_name=COLLECTION_NAME,
    connection=CONNECTION_STRING,
    use_jsonb=True,  # 메타데이터 JSONB 저장
)

# 3. 문서 청킹 및 임베딩 저장
def ingest_documents(file_path: str, metadata: dict):
    with open(file_path, 'r', encoding='utf-8') as f:
        raw_text = f.read()

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=50,
        separators=["\n\n", "\n", ".", "。", " "]
    )
    chunks = splitter.split_text(raw_text)

    documents = [
        Document(page_content=chunk, metadata={**metadata, "chunk_index": i})
        for i, chunk in enumerate(chunks)
    ]

    ids = vector_store.add_documents(documents)
    print(f"Ingested {len(ids)} chunks from {file_path}")
    return ids

# 4. RAG 체인 구성
def build_rag_chain(k: int = 5, score_threshold: float = 0.7):
    retriever = vector_store.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={"k": k, "score_threshold": score_threshold}
    )

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
        verbose=True
    )
    return qa_chain

# 5. 쿼리 실행
rag_chain = build_rag_chain()
result = rag_chain.invoke({"query": "PostgreSQL MVCC는 어떻게 동시성을 처리하나요?"})

print(result['result'])
for doc in result['source_documents']:
    print(f"Source: {doc.metadata.get('source')}, Chunk: {doc.metadata.get('chunk_index')}")

from sqlalchemy import text

def hybrid_search(
    query: str,
    query_embedding: list,
    k: int = 10,
    alpha: float = 0.5  # 0=BM25 only, 1=벡터 only
) -> list:
    """RRF (Reciprocal Rank Fusion)로 결과 결합"""

    sql = text("""
    WITH
    vector_search AS (
        SELECT id, content, metadata,
               ROW_NUMBER() OVER (ORDER BY embedding <=> :embedding) AS rank
        FROM document_embeddings
        ORDER BY embedding <=> :embedding
        LIMIT :k
    ),
    bm25_search AS (
        SELECT id, content, metadata,
               ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('korean', content),
                                  plainto_tsquery('korean', :query)) DESC) AS rank
        FROM document_embeddings
        WHERE to_tsvector('korean', content) @@ plainto_tsquery('korean', :query)
        LIMIT :k
    ),
    rrf_scores AS (
        SELECT
            COALESCE(v.id, b.id) AS id,
            COALESCE(v.content, b.content) AS content,
            COALESCE(v.metadata, b.metadata) AS metadata,
            COALESCE(1.0 / (60 + v.rank), 0) * :alpha +
            COALESCE(1.0 / (60 + b.rank), 0) * (1 - :alpha) AS rrf_score
        FROM vector_search v
        FULL OUTER JOIN bm25_search b ON v.id = b.id
    )
    SELECT id, content, metadata, rrf_score
    FROM rrf_scores
    ORDER BY rrf_score DESC
    LIMIT :k
    """)

    with engine.connect() as conn:
        results = conn.execute(sql, {
            "embedding": str(query_embedding),
            "query": query,
            "k": k,
            "alpha": alpha
        })
        return [dict(row) for row in results]

import hashlib
import json
from typing import Optional

class EmbeddingCache:
    def __init__(self, redis_client, ttl: int = 86400 * 7):  # 7일 캐시
        self.redis = redis_client
        self.ttl = ttl

    def _cache_key(self, text: str, model: str) -> str:
        content_hash = hashlib.sha256(f"{model}:{text}".encode()).hexdigest()
        return f"embedding:{content_hash}"

    def get(self, text: str, model: str) -> Optional[list]:
        key = self._cache_key(text, model)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None

    def set(self, text: str, model: str, embedding: list) -> None:
        key = self._cache_key(text, model)
        self.redis.setex(key, self.ttl, json.dumps(embedding))

    def get_or_compute(self, text: str, model: str, compute_fn) -> list:
        cached = self.get(text, model)
        if cached:
            return cached
        embedding = compute_fn(text)
        self.set(text, model, embedding)
        return embedding