# DB 쿼리 병렬화 (Query Parallelization) 완벽 가이드

> **목적**: Python asyncio와 SQLAlchemy를 활용한 DB 쿼리 병렬화의 이론부터 실무 적용까지
> **대상**: 비동기 프로그래밍 기초 지식이 있는 백엔드 개발자
> **환경**: Python 3.11+, SQLAlchemy 2.0+, FastAPI

---

## 목차

1. [이론적 배경](#1-이론적-배경)
2. [핵심 개념](#2-핵심-개념)
3. [설계 시 주의사항](#3-설계-시-주의사항)
4. [실무 시나리오 예제](#4-실무-시나리오-예제)
5. [성능 측정 및 모니터링](#5-성능-측정-및-모니터링)
6. [Best Practices](#6-best-practices)

---

## 1. 이론적 배경

### 1.1 동기 vs 비동기 실행

```
[순차 실행 - Sequential]
Query A ──────────▶ (100ms)
                    Query B ──────────▶ (100ms)
                                        Query C ──────────▶ (100ms)
총 소요시간: 300ms

[병렬 실행 - Parallel]
Query A ──────────▶ (100ms)
Query B ──────────▶ (100ms)
Query C ──────────▶ (100ms)
총 소요시간: ~100ms (가장 느린 쿼리 기준)
```

### 1.2 왜 병렬화가 필요한가?

1. **I/O 바운드 작업의 특성**
   - DB 쿼리는 네트워크 I/O가 대부분 (실제 CPU 작업은 짧음)
   - 대기 시간 동안 다른 작업을 수행할 수 있음

2. **응답 시간 단축**
   - N개의 독립적인 쿼리: `O(sum)` → `O(max)`
   - 사용자 경험 개선

3. **리소스 효율성**
   - 커넥션 풀을 효율적으로 활용
   - 서버 처리량(throughput) 증가

### 1.3 asyncio.gather()의 동작 원리

```python
import asyncio

async def main():
    # gather()는 모든 코루틴을 동시에 스케줄링
    results = await asyncio.gather(
        coroutine_1(),  # Task 1 생성
        coroutine_2(),  # Task 2 생성
        coroutine_3(),  # Task 3 생성
    )
    # 모든 Task가 완료되면 결과를 리스트로 반환
    return results
```

**핵심 동작:**
1. `gather()`는 각 코루틴을 Task로 래핑
2. 이벤트 루프가 모든 Task를 동시에 실행
3. I/O 대기 시 다른 Task로 컨텍스트 스위칭
4. 모든 Task 완료 시 결과 반환

---

## 2. 핵심 개념

### 2.1 독립성 판단 기준

병렬화가 가능한 쿼리의 조건:

| 조건 | 설명 | 예시 |
|------|------|------|
| **데이터 독립성** | 쿼리 간 결과 의존성 없음 | User, Product, Order 각각 조회 |
| **트랜잭션 독립성** | 같은 트랜잭션 내 순서 무관 | READ 작업들 |
| **비즈니스 독립성** | 결과 순서가 로직에 영향 없음 | 대시보드 데이터 조회 |

### 2.2 병렬화 불가능한 경우

```python
# ❌ 잘못된 예: 의존성이 있는 쿼리
user = await session.execute(select(User).where(User.id == user_id))
# orders 쿼리는 user.id에 의존 → 병렬화 불가
orders = await session.execute(
    select(Order).where(Order.user_id == user.id)
)
```

```python
# ❌ 잘못된 예: 쓰기 후 읽기 (Write-then-Read)
await session.execute(insert(User).values(name="John"))
# 방금 생성된 데이터를 조회 → 순차 실행 필요
new_user = await session.execute(select(User).where(User.name == "John"))
```

### 2.3 SQLAlchemy AsyncSession과 병렬 쿼리

**중요**: 하나의 AsyncSession 내에서 `asyncio.gather()`로 여러 쿼리를 실행할 수 있습니다.

```python
async with AsyncSessionLocal() as session:
    # 같은 세션에서 병렬 쿼리 실행 가능
    results = await asyncio.gather(
        session.execute(query1),
        session.execute(query2),
        session.execute(query3),
    )
```

**단, 주의사항:**
- 같은 세션은 같은 트랜잭션을 공유
- 하나의 쿼리 실패 시 전체 트랜잭션에 영향
- 커넥션 풀 크기 고려 필요

---

## 3. 설계 시 주의사항

### 3.1 커넥션 풀 크기 설정

```python
# SQLAlchemy 엔진 설정
engine = create_async_engine(
    url=db_url,
    pool_size=20,        # 기본 풀 크기
    max_overflow=20,     # 추가 연결 허용 수
    pool_timeout=30,     # 풀에서 연결 대기 시간
    pool_recycle=3600,   # 연결 재생성 주기
    pool_pre_ping=True,  # 연결 유효성 검사
)
```

**풀 크기 계산 공식:**
```
필요 커넥션 수 = 동시 요청 수 × 요청당 병렬 쿼리 수
```

예: 동시 10개 요청, 각 요청당 4개 병렬 쿼리
→ 최소 40개 커넥션 필요 (pool_size + max_overflow >= 40)

### 3.2 에러 처리 전략

```python
import asyncio

# 방법 1: return_exceptions=True (권장)
results = await asyncio.gather(
    session.execute(query1),
    session.execute(query2),
    session.execute(query3),
    return_exceptions=True,  # 예외를 결과로 반환
)

# 결과 처리
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Query {i} failed: {result}")
    else:
        print(f"Query {i} succeeded: {result}")
```

```python
# 방법 2: 개별 try-except 래핑
async def safe_execute(session, query, name: str):
    try:
        return await session.execute(query)
    except Exception as e:
        print(f"[{name}] Query failed: {e}")
        return None

results = await asyncio.gather(
    safe_execute(session, query1, "project"),
    safe_execute(session, query2, "song"),
    safe_execute(session, query3, "image"),
)
```

### 3.3 타임아웃 설정

```python
import asyncio

async def execute_with_timeout(session, query, timeout_seconds: float):
    """타임아웃이 있는 쿼리 실행"""
    try:
        return await asyncio.wait_for(
            session.execute(query),
            timeout=timeout_seconds
        )
    except asyncio.TimeoutError:
        raise Exception(f"Query timed out after {timeout_seconds}s")

# 사용 예
results = await asyncio.gather(
    execute_with_timeout(session, query1, 5.0),
    execute_with_timeout(session, query2, 5.0),
    execute_with_timeout(session, query3, 10.0),  # 더 긴 타임아웃
)
```

### 3.4 N+1 문제와 병렬화

```python
# ❌ N+1 문제 발생 코드
videos = await session.execute(select(Video))
for video in videos.scalars():
    # N번의 추가 쿼리 발생!
    project = await session.execute(
        select(Project).where(Project.id == video.project_id)
    )

# ✅ 해결 방법 1: JOIN 사용
query = select(Video).options(selectinload(Video.project))
videos = await session.execute(query)

# ✅ 해결 방법 2: IN 절로 배치 조회
video_list = videos.scalars().all()
project_ids = [v.project_id for v in video_list if v.project_id]

projects_result = await session.execute(
    select(Project).where(Project.id.in_(project_ids))
)
projects_map = {p.id: p for p in projects_result.scalars().all()}
```

### 3.5 트랜잭션 격리 수준 고려

| 격리 수준 | 병렬 쿼리 안전성 | 설명 |
|-----------|------------------|------|
| READ UNCOMMITTED | ⚠️ 주의 | Dirty Read 가능 |
| READ COMMITTED | ✅ 안전 | 대부분의 경우 적합 |
| REPEATABLE READ | ✅ 안전 | 일관된 스냅샷 |
| SERIALIZABLE | ✅ 안전 | 성능 저하 가능 |

---

## 4. 실무 시나리오 예제

### 4.1 시나리오 1: 대시보드 데이터 조회

**요구사항**: 사용자 대시보드에 필요한 여러 통계 데이터를 한 번에 조회

```python
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio


async def get_dashboard_data(
    session: AsyncSession,
    user_id: int,
) -> dict:
    """
    대시보드에 필요한 모든 데이터를 병렬로 조회합니다.

    조회 항목:
    - 사용자 정보
    - 최근 주문 5개
    - 총 주문 금액
    - 찜한 상품 수
    """

    # 1. 쿼리 정의 (아직 실행하지 않음)
    user_query = select(User).where(User.id == user_id)

    recent_orders_query = (
        select(Order)
        .where(Order.user_id == user_id)
        .order_by(Order.created_at.desc())
        .limit(5)
    )

    total_amount_query = (
        select(func.sum(Order.amount))
        .where(Order.user_id == user_id)
    )

    wishlist_count_query = (
        select(func.count(Wishlist.id))
        .where(Wishlist.user_id == user_id)
    )

    # 2. 4개 쿼리를 병렬로 실행
    user_result, orders_result, amount_result, wishlist_result = (
        await asyncio.gather(
            session.execute(user_query),
            session.execute(recent_orders_query),
            session.execute(total_amount_query),
            session.execute(wishlist_count_query),
        )
    )

    # 3. 결과 처리
    user = user_result.scalar_one_or_none()
    if not user:
        raise ValueError(f"User {user_id} not found")

    return {
        "user": {
            "id": user.id,
            "name": user.name,
            "email": user.email,
        },
        "recent_orders": [
            {"id": o.id, "amount": o.amount, "status": o.status}
            for o in orders_result.scalars().all()
        ],
        "total_spent": amount_result.scalar() or 0,
        "wishlist_count": wishlist_result.scalar() or 0,
    }


# 사용 예시 (FastAPI)
@router.get("/dashboard")
async def dashboard(
    user_id: int,
    session: AsyncSession = Depends(get_session),
):
    return await get_dashboard_data(session, user_id)
```

**성능 비교:**
- 순차 실행: ~200ms (50ms × 4)
- 병렬 실행: ~60ms (가장 느린 쿼리 기준)
- **개선율: 약 70%**

---

### 4.2 시나리오 2: 복합 검색 결과 조회

**요구사항**: 검색 결과와 함께 필터 옵션(카테고리 수, 가격 범위 등)을 조회

```python
from sqlalchemy import select, func, and_
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio
from typing import NamedTuple


class SearchFilters(NamedTuple):
    """검색 필터 결과"""
    categories: list[dict]
    price_range: dict
    brands: list[dict]


class SearchResult(NamedTuple):
    """전체 검색 결과"""
    items: list
    total_count: int
    filters: SearchFilters


async def search_products_with_filters(
    session: AsyncSession,
    keyword: str,
    page: int = 1,
    page_size: int = 20,
) -> SearchResult:
    """
    상품 검색과 필터 옵션을 병렬로 조회합니다.

    병렬 실행 쿼리:
    1. 상품 목록 (페이지네이션)
    2. 전체 개수
    3. 카테고리별 개수
    4. 가격 범위 (min, max)
    5. 브랜드별 개수
    """

    # 기본 검색 조건
    base_condition = Product.name.ilike(f"%{keyword}%")

    # 쿼리 정의
    items_query = (
        select(Product)
        .where(base_condition)
        .order_by(Product.created_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )

    count_query = (
        select(func.count(Product.id))
        .where(base_condition)
    )

    category_stats_query = (
        select(
            Product.category_id,
            Category.name.label("category_name"),
            func.count(Product.id).label("count")
        )
        .join(Category, Product.category_id == Category.id)
        .where(base_condition)
        .group_by(Product.category_id, Category.name)
    )

    price_range_query = (
        select(
            func.min(Product.price).label("min_price"),
            func.max(Product.price).label("max_price"),
        )
        .where(base_condition)
    )

    brand_stats_query = (
        select(
            Product.brand,
            func.count(Product.id).label("count")
        )
        .where(and_(base_condition, Product.brand.isnot(None)))
        .group_by(Product.brand)
        .order_by(func.count(Product.id).desc())
        .limit(10)
    )

    # 5개 쿼리 병렬 실행
    (
        items_result,
        count_result,
        category_result,
        price_result,
        brand_result,
    ) = await asyncio.gather(
        session.execute(items_query),
        session.execute(count_query),
        session.execute(category_stats_query),
        session.execute(price_range_query),
        session.execute(brand_stats_query),
    )

    # 결과 처리
    items = items_result.scalars().all()
    total_count = count_result.scalar() or 0

    categories = [
        {"id": row.category_id, "name": row.category_name, "count": row.count}
        for row in category_result.all()
    ]

    price_row = price_result.one()
    price_range = {
        "min": float(price_row.min_price or 0),
        "max": float(price_row.max_price or 0),
    }

    brands = [
        {"name": row.brand, "count": row.count}
        for row in brand_result.all()
    ]

    return SearchResult(
        items=items,
        total_count=total_count,
        filters=SearchFilters(
            categories=categories,
            price_range=price_range,
            brands=brands,
        ),
    )


# 사용 예시 (FastAPI)
@router.get("/search")
async def search(
    keyword: str,
    page: int = 1,
    session: AsyncSession = Depends(get_session),
):
    result = await search_products_with_filters(session, keyword, page)
    return {
        "items": [item.to_dict() for item in result.items],
        "total_count": result.total_count,
        "filters": {
            "categories": result.filters.categories,
            "price_range": result.filters.price_range,
            "brands": result.filters.brands,
        },
    }
```

**성능 비교:**
- 순차 실행: ~350ms (70ms × 5)
- 병렬 실행: ~80ms
- **개선율: 약 77%**

---

### 4.3 시나리오 3: 다중 테이블 데이터 수집 (본 프로젝트 실제 적용 예)

**요구사항**: 영상 생성을 위해 Project, Lyric, Song, Image 데이터를 한 번에 조회

```python
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio
from dataclasses import dataclass
from fastapi import HTTPException


@dataclass
class VideoGenerationData:
    """영상 생성에 필요한 데이터"""
    project_id: int
    lyric_id: int
    song_id: int
    music_url: str
    song_duration: float
    lyrics: str
    image_urls: list[str]


async def fetch_video_generation_data(
    session: AsyncSession,
    task_id: str,
) -> VideoGenerationData:
    """
    영상 생성에 필요한 모든 데이터를 병렬로 조회합니다.

    이 함수는 4개의 독립적인 테이블을 조회합니다:
    - Project: 프로젝트 정보
    - Lyric: 가사 정보
    - Song: 노래 정보 (음악 URL, 길이, 가사)
    - Image: 이미지 목록

    각 테이블은 task_id로 연결되어 있으며, 서로 의존성이 없으므로
    병렬 조회가 가능합니다.
    """

    # ============================================================
    # Step 1: 쿼리 객체 생성 (아직 실행하지 않음)
    # ============================================================
    project_query = (
        select(Project)
        .where(Project.task_id == task_id)
        .order_by(Project.created_at.desc())
        .limit(1)
    )

    lyric_query = (
        select(Lyric)
        .where(Lyric.task_id == task_id)
        .order_by(Lyric.created_at.desc())
        .limit(1)
    )

    song_query = (
        select(Song)
        .where(Song.task_id == task_id)
        .order_by(Song.created_at.desc())
        .limit(1)
    )

    image_query = (
        select(Image)
        .where(Image.task_id == task_id)
        .order_by(Image.img_order.asc())
    )

    # ============================================================
    # Step 2: asyncio.gather()로 4개 쿼리 병렬 실행
    # ============================================================
    #
    # 병렬 실행의 핵심:
    # - 각 쿼리는 독립적 (서로의 결과에 의존하지 않음)
    # - 같은 세션 내에서 실행 (같은 트랜잭션 공유)
    # - 가장 느린 쿼리 시간만큼만 소요됨
    #
    project_result, lyric_result, song_result, image_result = (
        await asyncio.gather(
            session.execute(project_query),
            session.execute(lyric_query),
            session.execute(song_query),
            session.execute(image_query),
        )
    )

    # ============================================================
    # Step 3: 결과 검증 및 데이터 추출
    # ============================================================

    # Project 검증
    project = project_result.scalar_one_or_none()
    if not project:
        raise HTTPException(
            status_code=404,
            detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
        )

    # Lyric 검증
    lyric = lyric_result.scalar_one_or_none()
    if not lyric:
        raise HTTPException(
            status_code=404,
            detail=f"task_id '{task_id}'에 해당하는 Lyric을 찾을 수 없습니다.",
        )

    # Song 검증 및 데이터 추출
    song = song_result.scalar_one_or_none()
    if not song:
        raise HTTPException(
            status_code=404,
            detail=f"task_id '{task_id}'에 해당하는 Song을 찾을 수 없습니다.",
        )

    if not song.song_result_url:
        raise HTTPException(
            status_code=400,
            detail=f"Song(id={song.id})의 음악 URL이 없습니다.",
        )

    if not song.song_prompt:
        raise HTTPException(
            status_code=400,
            detail=f"Song(id={song.id})의 가사(song_prompt)가 없습니다.",
        )

    # Image 검증
    images = image_result.scalars().all()
    if not images:
        raise HTTPException(
            status_code=404,
            detail=f"task_id '{task_id}'에 해당하는 이미지를 찾을 수 없습니다.",
        )

    # ============================================================
    # Step 4: 결과 반환
    # ============================================================
    return VideoGenerationData(
        project_id=project.id,
        lyric_id=lyric.id,
        song_id=song.id,
        music_url=song.song_result_url,
        song_duration=song.duration or 60.0,
        lyrics=song.song_prompt,
        image_urls=[img.img_url for img in images],
    )


# 실제 사용 예시
async def generate_video(task_id: str) -> dict:
    async with AsyncSessionLocal() as session:
        # 병렬 쿼리로 데이터 조회
        data = await fetch_video_generation_data(session, task_id)

        # Video 레코드 생성
        video = Video(
            project_id=data.project_id,
            lyric_id=data.lyric_id,
            song_id=data.song_id,
            task_id=task_id,
            status="processing",
        )
        session.add(video)
        await session.commit()

    # 세션 종료 후 외부 API 호출
    # (커넥션 타임아웃 방지)
    return await call_creatomate_api(data)
```

**성능 비교:**
- 순차 실행: ~200ms (약 50ms × 4쿼리)
- 병렬 실행: ~55ms
- **개선율: 약 72%**

---

## 5. 성능 측정 및 모니터링

### 5.1 실행 시간 측정 데코레이터

```python
import time
import functools
from typing import Callable, TypeVar

T = TypeVar("T")


def measure_time(func: Callable[..., T]) -> Callable[..., T]:
    """함수 실행 시간을 측정하는 데코레이터"""

    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return await func(*args, **kwargs)
        finally:
            elapsed = (time.perf_counter() - start) * 1000
            print(f"[{func.__name__}] Execution time: {elapsed:.2f}ms")

    @functools.wraps(func)
    def sync_wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            elapsed = (time.perf_counter() - start) * 1000
            print(f"[{func.__name__}] Execution time: {elapsed:.2f}ms")

    if asyncio.iscoroutinefunction(func):
        return async_wrapper
    return sync_wrapper


# 사용 예
@measure_time
async def fetch_data(session, task_id):
    ...
```

### 5.2 병렬 쿼리 성능 비교 유틸리티

```python
import asyncio
import time


async def compare_sequential_vs_parallel(
    session: AsyncSession,
    queries: list,
    labels: list[str] | None = None,
) -> dict:
    """순차 실행과 병렬 실행의 성능을 비교합니다."""

    labels = labels or [f"Query {i}" for i in range(len(queries))]

    # 순차 실행
    sequential_start = time.perf_counter()
    sequential_results = []
    for query in queries:
        result = await session.execute(query)
        sequential_results.append(result)
    sequential_time = (time.perf_counter() - sequential_start) * 1000

    # 병렬 실행
    parallel_start = time.perf_counter()
    parallel_results = await asyncio.gather(
        *[session.execute(query) for query in queries]
    )
    parallel_time = (time.perf_counter() - parallel_start) * 1000

    improvement = ((sequential_time - parallel_time) / sequential_time) * 100

    return {
        "sequential_time_ms": round(sequential_time, 2),
        "parallel_time_ms": round(parallel_time, 2),
        "improvement_percent": round(improvement, 1),
        "query_count": len(queries),
    }
```

### 5.3 SQLAlchemy 쿼리 로깅

```python
import logging

# SQLAlchemy 쿼리 로깅 활성화
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# 또는 엔진 생성 시 echo=True
engine = create_async_engine(url, echo=True)
```

---

## 6. Best Practices

### 6.1 체크리스트

병렬화 적용 전 확인사항:

- [ ] 쿼리들이 서로 독립적인가? (결과 의존성 없음)
- [ ] 모든 쿼리가 READ 작업인가? (또는 순서 무관한 WRITE)
- [ ] 커넥션 풀 크기가 충분한가?
- [ ] 에러 처리 전략이 수립되어 있는가?
- [ ] 타임아웃 설정이 적절한가?

### 6.2 권장 패턴

```python
# ✅ 권장: 쿼리 정의와 실행 분리
async def fetch_data(session: AsyncSession, task_id: str):
    # 1. 쿼리 객체 정의 (명확한 의도 표현)
    project_query = select(Project).where(Project.task_id == task_id)
    song_query = select(Song).where(Song.task_id == task_id)

    # 2. 병렬 실행
    results = await asyncio.gather(
        session.execute(project_query),
        session.execute(song_query),
    )

    # 3. 결과 처리
    return process_results(results)
```

### 6.3 피해야 할 패턴

```python
# ❌ 피하기: 인라인 쿼리 (가독성 저하)
results = await asyncio.gather(
    session.execute(select(A).where(A.x == y).order_by(A.z.desc()).limit(1)),
    session.execute(select(B).where(B.a == b).order_by(B.c.desc()).limit(1)),
)

# ❌ 피하기: 과도한 병렬화 (커넥션 고갈)
# 100개 쿼리를 동시에 실행하면 커넥션 풀 고갈 위험
results = await asyncio.gather(*[session.execute(q) for q in queries])

# ✅ 해결: 배치 처리
BATCH_SIZE = 10
for i in range(0, len(queries), BATCH_SIZE):
    batch = queries[i:i + BATCH_SIZE]
    results = await asyncio.gather(*[session.execute(q) for q in batch])
```

### 6.4 성능 최적화 팁

1. **인덱스 확인**: 병렬화해도 인덱스 없으면 느림
2. **쿼리 최적화 우선**: 병렬화 전에 개별 쿼리 최적화
3. **적절한 병렬 수준**: 보통 3-10개가 적절
4. **모니터링 필수**: 실제 개선 효과 측정

---

## 부록: 관련 자료

- [SQLAlchemy 2.0 AsyncIO 문서](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
- [Python asyncio 공식 문서](https://docs.python.org/3/library/asyncio.html)
- [FastAPI 비동기 데이터베이스](https://fastapi.tiangolo.com/async/)