Какие знаешь методы оптимизации запросов баз данных?

Методы оптимизации запросов баз данных

Оптимизация запросов — критична для производительности приложения. Медленный запрос может привести к простоям сервера, даже если код идеален.

1. Использование индексов

Основной инструмент оптимизации — B-Tree индексы

# Без индекса: O(n) — полный scan таблицы
SELECT * FROM users WHERE email = 'user@example.com'
# Проверяет все 1,000,000 строк

# С индексом на email: O(log n) — поиск в B-Tree
CREATE INDEX idx_users_email ON users(email)
SELECT * FROM users WHERE email = 'user@example.com'
# Находит за ~20 операций

Типы индексов:

-- Single-column индекс (самый частый)
CREATE INDEX idx_users_email ON users(email);

-- Composite индекс (для запросов с несколькими условиями)
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at);

-- UNIQUE индекс (гарантирует уникальность)
CREATE UNIQUE INDEX idx_users_username ON users(username);

-- Partial индекс (только для части строк)
CREATE INDEX idx_active_users ON users(id) WHERE active = true;

-- Expression индекс (по вычисляемым значениям)
CREATE INDEX idx_users_name_lower ON users(LOWER(name));

Правило: индексируй колонки в WHERE, JOIN, ORDER BY

2. Избегай N+1 запросов (Eager Loading)

N+1 проблема — запрос за каждой строкой

# ❌ ПЛОХО: N+1 запросов (1 + 1000)
users = User.query.all()  # 1 запрос
for user in users:  # 1000 users
    print(user.posts)  # 1000 дополнительных запросов!

# ✅ ХОРОШО: Eager loading с JOIN (1 запрос)
from sqlalchemy.orm import joinedload
users = User.query.options(joinedload(User.posts)).all()
for user in users:
    print(user.posts)  # Данные уже загружены

# ✅ ХОРОШО: Explicit JOIN
from sqlalchemy import select
stmt = select(User).join(Post).options(joinedload(User.posts))
users = session.execute(stmt).unique().scalars()

3. Выбирай только нужные колонки

Передача лишних данных — пустая траата памяти и I/O

# ❌ ПЛОХО: выбираем все (включая большие текстовые поля)
SELECT * FROM users
# Берём id, name, email, bio, profile_pic_blob...

# ✅ ХОРОШО: выбираем только нужное
SELECT id, name, email FROM users

# В Python/SQLAlchemy:
from sqlalchemy import select
stmt = select(User.id, User.name, User.email)
users = session.execute(stmt).all()

4. Используй EXPLAIN для анализа

Посмотри, как база выполняет запрос

-- PostgreSQL
EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM users WHERE email = 'user@example.com';

Output:
Seq Scan on users (cost=0.00..35.50 rows=1 width=100)
  Filter: (email = 'user@example.com')
  -> много информации о том, как выполняется запрос

-- Если видишь "Seq Scan" (Sequential Scan) вместо "Index Scan"
-- значит нет индекса или он не используется

5. Агрегация данных на стороне БД

Не тягай все данные в приложение, агрегируй в БД

# ❌ ПЛОХО: считаем в Python
all_orders = Order.query.filter(Order.user_id == user_id)
total = sum(order.amount for order in all_orders)
# Получаем все 10,000 заказов, потом считаем в памяти

# ✅ ХОРОШО: считаем в БД (O(n) на диске, не в памяти)
from sqlalchemy import func
from sqlalchemy import select

stmt = select(func.sum(Order.amount)).where(Order.user_id == user_id)
total = session.execute(stmt).scalar()
# База вычисляет сумму за O(n log n) с индексом

6. Кеширование результатов

Часто запрашиваемые данные кешируй в Redis

from redis import Redis
import json

redis = Redis()

def get_user_profile(user_id):
    # Пытаемся получить из кеша
    cached = redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # Если нет — запрашиваем из БД
    user = User.query.get(user_id)
    
    # Сохраняем в кеш на 1 час
    redis.setex(
        f"user:{user_id}",
        3600,
        json.dumps(user.to_dict())
    )
    
    return user

7. Batch обработка вместо однократных операций

Вставляй сотни строк одним INSERT, а не тысячей запросов

# ❌ ПЛОХО: 1000 INSERT запросов
for item in items:
    db.execute(
        "INSERT INTO items (name, price) VALUES (?, ?)",
        (item['name'], item['price'])
    )
db.commit()  # 1000 транзакций!

# ✅ ХОРОШО: один INSERT с множеством значений
values = [(item['name'], item['price']) for item in items]
db.execute(
    "INSERT INTO items (name, price) VALUES (?, ?)",
    values
)
db.commit()  # 1 транзакция

8. Используй LIMIT при разработке

Не выбирай 1,000,000 строк, когда нужны первые 10

# ❌ ПЛОХО: выбираем все для пагинации
users = User.query.all()  # 1 млн строк
page_users = users[(page-1)*10 : page*10]

# ✅ ХОРОШО: LIMIT + OFFSET в БД
users = User.query.limit(10).offset((page-1)*10).all()

9. Денормализация для читаемых запросов

Иногда дублирование данных быстрее, чем JOIN

# Нормализовано (медленнее):
SELECT o.id, o.date, u.name, u.email
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.id = 123;

-- 2 таблицы, JOIN

# Денормализовано (быстрее):
SELECT id, date, user_name, user_email
FROM orders
WHERE id = 123;

-- 1 таблица, никаких JOIN
-- Храним user_name и user_email в orders
-- Обновляем при изменении имени пользователя

10. Используй правильные типы данных

Целое число быстрее, чем строка

-- ❌ ПЛОХО: status как VARCHAR(50)
CREATE TABLE orders (
    id INT,
    status VARCHAR(50)  -- Каждый поиск = сравнение строк
);

-- ✅ ХОРОШО: status как ENUM или INT
CREATE TABLE orders (
    id INT,
    status ENUM('pending', 'completed', 'cancelled')
);

-- ENUM занимает 1 байт вместо 50, поиск в 50 раз быстрее

11. Используй транзакции для групповых операций

from sqlalchemy import select

# Группируем несколько операций
with session.begin():
    user = User(name="Alice", email="alice@example.com")
    session.add(user)
    session.flush()  # Получаем user.id
    
    post = Post(title="Hello", user_id=user.id)
    session.add(post)
    # При выходе из блока — COMMIT

# Или ROLLBACK если ошибка

12. Профилирование и мониторинг

import time
from sqlalchemy import event, Engine

@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total_time = time.time() - conn.info['query_start_time'].pop(-1)
    print(f"Query took {total_time:.4f}s: {statement}")

# Теперь видишь время каждого запроса!

Чек-лист оптимизации

1. Выбери только нужные колонки (не SELECT *)
2. Добавь индексы на колонки в WHERE, JOIN, ORDER BY
3. Избегай N+1 (используй eager loading)
4. Агрегируй в БД (COUNT, SUM, GROUP BY)
5. Кешируй результаты (Redis, memcached)
6. Используй LIMIT для пагинации
7. Пиши EXPLAIN перед оптимизацией
8. Batch вставки/обновления
9. Денормализуй, если нужна скорость
10. Профилируй запросы (найди узкие места)

Быстрые факты

• Индекс может сделать запрос в 100 раз быстрее
• N+1 проблема может замедлить на 1000x
• Кеширование может сделать на 10,000x быстрее
• EXPLAIN должен быть твоим лучшим другом

Оптимизация БД — это 80% производительности приложения!