N+1 問題を完全解剖 — ORM の静かなパフォーマンスキラー

# モデル定義（Django）
class Author(models.Model):
    name = models.CharField(max_length=100)

class Post(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(Author, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

# N+1 問題発生！
posts = Post.objects.all()[:100]  # 1回目のクエリ：記事100件を取得

for post in posts:
    print(f"{post.title} by {post.author.name}")
    # ↑ 記事ごとに author を照会！100回の追加クエリ！

-- 1回目のクエリ（記事一覧）
SELECT id, title, author_id, created_at FROM posts LIMIT 100;

-- 2回目～101回目のクエリ（著者を個別に照会！）
SELECT id, name FROM authors WHERE id = 1;
SELECT id, name FROM authors WHERE id = 2;
SELECT id, name FROM authors WHERE id = 3;
SELECT id, name FROM authors WHERE id = 1;  -- 同じ著者も再度照会！
SELECT id, name FROM authors WHERE id = 4;
...
-- 合計101回のクエリ！（1 + N = 1 + 100）

# ベンチマーク
import time

# N+1（101クエリ）
start = time.time()
posts = Post.objects.all()[:1000]
for post in posts:
    _ = post.author.name
print(f"N+1: {time.time() - start:.2f}s")
# → N+1: 3.47s（1001クエリ）

# 最適化後（2クエリ）
start = time.time()
posts = Post.objects.select_related('author').all()[:1000]
for post in posts:
    _ = post.author.name
print(f"Optimized: {time.time() - start:.2f}s")
# → Optimized: 0.05s（2クエリ）

# 69倍高速！！

データ数    │  N+1 クエリ数  │  応答時間      │  最適化後
──────────┼──────────────┼─────────────┼──────────
10        │  11          │  0.05s      │  0.01s
100       │  101         │  0.35s      │  0.02s
1,000     │  1,001       │  3.5s       │  0.05s
10,000    │  10,001      │  35s        │  0.1s
100,000   │  100,001     │  5分+       │  0.3s

# Post → Author（Many-to-One）
# 記事から著者を照会

# N+1
for post in Post.objects.all():
    print(post.author.name)  # 毎回 SELECT

# select_related（SQL JOIN）
for post in Post.objects.select_related('author'):
    print(post.author.name)  # JOIN で一度に取得！

-- select_related → INNER JOIN
SELECT posts.*, authors.name
FROM posts
INNER JOIN authors ON posts.author_id = authors.id
LIMIT 100;
-- 1回のクエリ！

# Author → Posts（One-to-Many）
# 著者別の記事一覧を照会

# N+1
authors = Author.objects.all()[:50]
for author in authors:
    posts = author.post_set.all()  # 毎回 SELECT
    print(f"{author.name}: {posts.count()} posts")

# prefetch_related（別クエリ + Python でマッピング）
authors = Author.objects.prefetch_related('post_set').all()[:50]
for author in authors:
    posts = author.post_set.all()  # キャッシュされた結果を使用！
    print(f"{author.name}: {len(posts)} posts")

-- prefetch_related → 2回のクエリ
SELECT id, name FROM authors LIMIT 50;
SELECT id, title, author_id FROM posts WHERE author_id IN (1, 2, 3, ..., 50);
-- Python でマッピング！

class Post(models.Model):
    tags = models.ManyToManyField('Tag')

# N+1
for post in Post.objects.all():
    tag_names = [t.name for t in post.tags.all()]  # 毎回 SELECT

# prefetch_related
for post in Post.objects.prefetch_related('tags'):
    tag_names = [t.name for t in post.tags.all()]  # キャッシュ！

# Post → Author → Company（3段階ネスト）

# N+1+1（記事 N + 著者 N + 会社 N）
for post in Post.objects.all():
    print(f"{post.title} by {post.author.name} at {post.author.company.name}")
    # 記事ごとに: author SELECT + company SELECT

# select_related チェイニング
for post in Post.objects.select_related('author__company'):
    print(f"{post.title} by {post.author.name} at {post.author.company.name}")

-- 3重 JOIN、1回のクエリ！
SELECT posts.*, authors.*, companies.*
FROM posts
JOIN authors ON posts.author_id = authors.id
JOIN companies ON authors.company_id = companies.id;

select_related:
  ├── SQL JOIN で一度に取得
  ├── ForeignKey / OneToOneField にのみ使用
  ├── 1回のクエリ（JOIN）
  └── データが少ないときに効率的

prefetch_related:
  ├── 別クエリ + Python でマッピング
  ├── ManyToMany / Reverse FK に使用
  ├── 2回のクエリ（メイン + IN）
  └── データが多いか複雑なリレーションで効率的

# 両方同時に使用可能！
posts = Post.objects.select_related(
    'author',           # FK → JOIN
    'author__company',  # FK → JOIN
).prefetch_related(
    'tags',             # M2M → 別クエリ
    'comments',         # Reverse FK → 別クエリ
).all()[:100]

# たった4回のクエリですべてのリレーションをロード！

# select_related（FK, O2O）
Post.objects.select_related('author', 'category')

# prefetch_related（M2M, Reverse FK）
Post.objects.prefetch_related('tags', 'comments')

# Prefetch オブジェクト（フィルタリング + 最適化）
from django.db.models import Prefetch

Post.objects.prefetch_related(
    Prefetch(
        'comments',
        queryset=Comment.objects.filter(is_approved=True).select_related('user'),
        to_attr='approved_comments'  # post.approved_comments でアクセス
    )
)

# annotate で COUNT 最適化（サブクエリの代わりに集計）
from django.db.models import Count
Author.objects.annotate(post_count=Count('post')).filter(post_count__gte=5)

from sqlalchemy.orm import joinedload, selectinload, subqueryload

# joinedload = select_related（JOIN）
session.query(Post).options(joinedload(Post.author)).all()

# selectinload = prefetch_related（SELECT ... IN）
session.query(Author).options(selectinload(Author.posts)).all()

# ネスト
session.query(Post).options(
    joinedload(Post.author).joinedload(Author.company),
    selectinload(Post.tags)
).all()

# リレーションローディング戦略（モデルで設定）
class Post(Base):
    author = relationship("Author", lazy="joined")     # 常に JOIN
    tags = relationship("Tag", lazy="selectin")         # 常に IN クエリ
    comments = relationship("Comment", lazy="dynamic")  # 明示的なクエリのみ

// Lazy Loading → デフォルトで N+1 が発生！
@Entity
public class Post {
    @ManyToOne(fetch = FetchType.LAZY)  // デフォルトが LAZY
    private Author author;
}

// 解決法 1: JPQL JOIN FETCH
@Query("SELECT p FROM Post p JOIN FETCH p.author")
List<Post> findAllWithAuthor();

// 解決法 2: EntityGraph
@EntityGraph(attributePaths = {"author", "tags"})
List<Post> findAll();

// 解決法 3: Batch Size（Hibernate）
@BatchSize(size = 100)  // IN 句で100件ずつまとめて照会
@OneToMany(mappedBy = "author")
private List<Post> posts;

// application.yml
// spring.jpa.properties.hibernate.default_batch_fetch_size: 100

// N+1
const posts = await prisma.post.findMany()
for (const post of posts) {
  const author = await prisma.author.findUnique({
    where: { id: post.authorId },
  })
}

// include（Eager Loading）
const posts = await prisma.post.findMany({
  include: {
    author: true,
    tags: true,
    comments: {
      where: { approved: true },
      include: { user: true },
    },
  },
})

# N+1
Post.all.each { |post| puts post.author.name }

# includes（自動的に最適な戦略を選択）
Post.includes(:author).each { |post| puts post.author.name }

# eager_load（常に JOIN）
Post.eager_load(:author, :tags).all

# preload（常に別クエリ）
Post.preload(:comments).all

# Bullet gem で自動検出！
# Gemfile: gem 'bullet'

# settings.py
INSTALLED_APPS = ['debug_toolbar']
MIDDLEWARE = ['debug_toolbar.middleware.DebugToolbarMiddleware']

# → ブラウザで SQL クエリ数 + 重複を確認！

# Django: クエリ数自動ロギングミドルウェア
from django.db import connection, reset_queries
import logging

logger = logging.getLogger(__name__)

class QueryCountMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        reset_queries()
        response = self.get_response(request)
        query_count = len(connection.queries)

        if query_count > 10:  # 閾値
            logger.warning(
                f"警告: {request.path}: {query_count} queries! "
                f"Possible N+1 problem."
            )
            # 重複クエリを検出
            queries = [q['sql'] for q in connection.queries]
            from collections import Counter
            dupes = {q: c for q, c in Counter(queries).items() if c > 1}
            if dupes:
                logger.warning(f"Duplicate queries: {dupes}")

        return response

import pytest
from django.test.utils import override_settings

@pytest.mark.django_db
def test_no_n_plus_1_on_post_list(client, django_assert_max_num_queries):
    # 記事100件を作成
    author = Author.objects.create(name="Test")
    for i in range(100):
        Post.objects.create(title=f"Post {i}", author=author)

    # クエリ数制限（100件の記事で10回以下であるべき）
    with django_assert_max_num_queries(10):
        response = client.get('/api/posts/')
        assert response.status_code == 200

from sqlalchemy import event
import warnings

@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, *args):
    if not hasattr(conn, '_query_count'):
        conn._query_count = 0
    conn._query_count += 1
    if conn._query_count > 50:
        warnings.warn(f"Over 50 queries in one request! ({conn._query_count})")

# GraphQL は N+1 の温床！
# resolver がフィールドごとに呼び出されるため

# N+1 発生
class PostType:
    def resolve_author(post, info):
        return Author.objects.get(id=post.author_id)  # 毎回 SELECT！

# DataLoader でバッチ処理
from promise import Promise
from promise.dataloader import DataLoader

class AuthorLoader(DataLoader):
    def batch_load_fn(self, author_ids):
        # 一度にすべて取得！
        authors = {a.id: a for a in Author.objects.filter(id__in=author_ids)}
        return Promise.resolve([authors.get(id) for id in author_ids])

# リクエストごとに新しい loader を生成
author_loader = AuthorLoader()

class PostType:
    def resolve_author(post, info):
        return author_loader.load(post.author_id)  # バッチ処理！

リレーションタイプは？
├── ForeignKey / OneToOne
│   └── select_related（JOIN）
│
├── ManyToMany / Reverse FK
│   └── prefetch_related（IN クエリ）
│
├── ネストされたリレーション（A → B → C）
│   └── select_related チェイニング（'b__c'）
│
├── 条件付きプリフェッチ
│   └── Prefetch オブジェクト + queryset フィルタ
│
└── COUNT/SUM のみ必要
    └── annotate + 集計関数（プリフェッチより効率的）