N+1 Problem Deep Dive — The Silent Performance Killer in ORMs

# Model definition (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 problem occurs!
posts = Post.objects.all()[:100]  # Query 1: Fetch 100 posts

for post in posts:
    print(f"{post.title} by {post.author.name}")
    # Each post triggers a separate author query! 100 additional queries!

-- 1st query (post list)
SELECT id, title, author_id, created_at FROM posts LIMIT 100;

-- 2nd to 101st queries (each author queried individually!)
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;  -- Same author queried again!
SELECT id, name FROM authors WHERE id = 4;
...
-- Total 101 queries! (1 + N = 1 + 100)

# Benchmark
import time

# N+1 (101 queries)
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 queries)

# After optimization (2 queries)
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 queries)

# 69x faster!!

Records   |  N+1 Queries  |  Response Time  |  After Optimization
----------+--------------+-----------------+--------------------
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     |  5min+          |  0.3s

# Post -> Author (Many-to-One)
# Querying author from post

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

# select_related (SQL JOIN)
for post in Post.objects.select_related('author'):
    print(post.author.name)  # Fetched with JOIN at once!

-- select_related -> INNER JOIN
SELECT posts.*, authors.name
FROM posts
INNER JOIN authors ON posts.author_id = authors.id
LIMIT 100;
-- 1 query!

# Author -> Posts (One-to-Many)
# Querying posts per author

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

# prefetch_related (separate query + Python mapping)
authors = Author.objects.prefetch_related('post_set').all()[:50]
for author in authors:
    posts = author.post_set.all()  # Uses cached result!
    print(f"{author.name}: {len(posts)} posts")

-- prefetch_related -> 2 queries
SELECT id, name FROM authors LIMIT 50;
SELECT id, title, author_id FROM posts WHERE author_id IN (1, 2, 3, ..., 50);
-- Mapped in 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 every time

# prefetch_related
for post in Post.objects.prefetch_related('tags'):
    tag_names = [t.name for t in post.tags.all()]  # Cached!

# Post -> Author -> Company (3-level nesting)

# N+1+1 (N posts + N authors + N companies)
for post in Post.objects.all():
    print(f"{post.title} by {post.author.name} at {post.author.company.name}")
    # Per post: author SELECT + company SELECT

# select_related chaining
for post in Post.objects.select_related('author__company'):
    print(f"{post.title} by {post.author.name} at {post.author.company.name}")

-- Triple JOIN, 1 query!
SELECT posts.*, authors.*, companies.*
FROM posts
JOIN authors ON posts.author_id = authors.id
JOIN companies ON authors.company_id = companies.id;

select_related:
  |- Fetches everything at once with SQL JOIN
  |- Only for ForeignKey / OneToOneField
  |- 1 query (JOIN)
  +- Efficient when data is small

prefetch_related:
  |- Separate query + mapping in Python
  |- For ManyToMany / Reverse FK
  |- 2 queries (main + IN)
  +- Efficient for large or complex relationships

# Both can be used together!
posts = Post.objects.select_related(
    'author',           # FK -> JOIN
    'author__company',  # FK -> JOIN
).prefetch_related(
    'tags',             # M2M -> separate query
    'comments',         # Reverse FK -> separate query
).all()[:100]

# All relationships loaded in just 4 queries!

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

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

# Prefetch object (filtering + optimization)
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'  # Access via post.approved_comments
    )
)

# annotate for COUNT optimization (aggregation instead of subquery)
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()

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

# Relationship loading strategy (set at model level)
class Post(Base):
    author = relationship("Author", lazy="joined")     # Always JOIN
    tags = relationship("Tag", lazy="selectin")         # Always IN query
    comments = relationship("Comment", lazy="dynamic")  # Explicit query only

// Lazy Loading -> N+1 by default!
@Entity
public class Post {
    @ManyToOne(fetch = FetchType.LAZY)  // LAZY is the default
    private Author author;
}

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

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

// Solution 3: Batch Size (Hibernate)
@BatchSize(size = 100)  // Batch fetch 100 at a time with IN clause
@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 (automatically selects optimal strategy)
Post.includes(:author).each { |post| puts post.author.name }

# eager_load (always JOIN)
Post.eager_load(:author, :tags).all

# preload (always separate queries)
Post.preload(:comments).all

# Auto-detect with Bullet gem!
# Gemfile: gem 'bullet'

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

# -> Check SQL query count + duplicates in the browser!

# Django: Middleware for automatic query count logging
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:  # Threshold
            logger.warning(
                f"Warning: {request.path}: {query_count} queries! "
                f"Possible N+1 problem."
            )
            # Find duplicate queries
            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):
    # Create 100 posts
    author = Author.objects.create(name="Test")
    for i in range(100):
        Post.objects.create(title=f"Post {i}", author=author)

    # Limit query count (100 posts should need 10 or fewer queries)
    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 is a breeding ground for N+1!
# Resolvers are called per field

# N+1 occurs
class PostType:
    def resolve_author(post, info):
        return Author.objects.get(id=post.author_id)  # SELECT every time!

# Batching with DataLoader
from promise import Promise
from promise.dataloader import DataLoader

class AuthorLoader(DataLoader):
    def batch_load_fn(self, author_ids):
        # Fetch all at once!
        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])

# Create a new loader per request
author_loader = AuthorLoader()

class PostType:
    def resolve_author(post, info):
        return author_loader.load(post.author_id)  # Batched!

What is the relationship type?
|- ForeignKey / OneToOne
|  +- select_related (JOIN)
|
|- ManyToMany / Reverse FK
|  +- prefetch_related (IN query)
|
|- Nested relationship (A -> B -> C)
|  +- select_related chaining ('b__c')
|
|- Conditional prefetch
|  +- Prefetch object + queryset filter
|
+- Only need COUNT/SUM
   +- annotate + aggregate functions (more efficient than prefetch)