Промпт: миграция базы данных без простоя

Общие принципы подхода

Мы планируем изменить схему базы данных следующим образом:

• Разделим full_name на два поля (first_name, last_name), заполнив существующие данные.
• Сделаем поле phone обязательным и добавим уникальный индекс по этому полю.

Изменение схемы должно происходить плавно, чтобы минимизировать влияние на производительность системы и исключить простой приложения во время выполнения операций.

Используем подход Expand and Contract — сначала расширяем схему, вносим минимальные изменения, потом постепенно добавляем бизнес-логику и финализируем структуру.

---

Пошаговый план изменений (Expand & Contract)

Шаг 1: Expand — расширение структуры

Создаём новые столбцы и начинаем заполнять существующий набор данных небольшими партиями.

Миграция №1: создание новых полей и индексация телефона (по частям):

def upgrade():
    op.add_column('users', sa.Column('first_name', sa.String(length=255)))
    op.add_column('users', sa.Column('last_name', sa.String(length=255)))

    # Создаем index_concurrently для уникального индекса, который будет выполняться параллельно и асинхронно,
    # что позволяет избежать длительных блокировок.
    conn = op.get_bind()
    conn.execute(sa.text("CREATE UNIQUE INDEX CONCURRENTLY idx_users_phone_unique ON users(phone);"))

def downgrade():
    with op.batch_alter_table('users') as batch_op:
        batch_op.drop_index('idx_users_phone_unique')
        batch_op.drop_column('first_name')
        batch_op.drop_column('last_name')

После успешной миграции делаем следующий шаг.

Шаг 2: Заполнение новыми значениями batched-запросами (скрипт Python).

Заполняем таблицу маленькими порциями, чтобы не создавать большие транзакции и тяжелые блокировки.

import psycopg2
from time import sleep

# Параметры подключения
DB_HOST = 'your_db_host'
DB_NAME = 'your_db_name'
DB_USER = 'your_user'
DB_PASSWORD = 'your_password'
BATCH_SIZE = 10_000
PAUSE_SECONDS = 10

conn = psycopg2.connect(
    host=DB_HOST,
    database=DB_NAME,
    user=DB_USER,
    password=DB_PASSWORD
)
cur = conn.cursor()

query_template = """
UPDATE users 
SET first_name = split_part(full_name, ' ', 1),
    last_name = CASE WHEN strpos(full_name, ' ') > 0 THEN substring(full_name from position(' ' in full_name)+1 for char_length(full_name)) ELSE '' END
WHERE id BETWEEN %s AND %s;
"""

min_id = cur.fetchone()[0]
max_id = cur.fetchone()[0]

for start_id in range(min_id, max_id + BATCH_SIZE, BATCH_SIZE):
    end_id = min(start_id + BATCH_SIZE - 1, max_id)
    print(f'Processing IDs {start_id}-{end_id}')
    cur.execute(query_template, (start_id, end_id))
    conn.commit()
    sleep(PAUSE_SECONDS)

print('Migration completed.')
cur.close()
conn.close()

Размер партии в 10 тыс. записей достаточно мал, чтобы обновление прошло быстро и эффективно. Пауза между пакетами поможет разгрузить систему, особенно при высокой нагрузке.

Проверяем успешность миграции данных перед переходом к следующему этапу.

Шаг 3: Добавляем обязательность для поля phone

Создаем новую миграцию, где сделаем поле телефон обязательным (не допускающим пустых значений). Поскольку изначально в таблице около 3% null-значений, заполняем их заранее случайным временным значением до полной замены данных.

Миграция №2: заполнить пустые телефоны временными значениями и сделать обязательное поле:

def upgrade():
    # Временно заменяем NULL-телефоны на временное значение ("N/A"), которое позже удалится
    conn = op.get_bind()
    conn.execute(sa.text("UPDATE users SET phone='N/A' WHERE phone IS NULL;"))
    
    # Устанавливаем NOT NULL ограничение
    with op.batch_alter_table('users') as batch_op:
        batch_op.alter_column('phone', nullable=False)

def downgrade():
    with op.batch_alter_table('users') as batch_op:
        batch_op.alter_column('phone', nullable=True)
    conn = op.get_bind()
    conn.execute(sa.text("UPDATE users SET phone=NULL WHERE phone='N/A';"))

Переносите эту миграцию отдельно и убедитесь, что временные значения правильно установлены.

Шаг 4: Code Change — модификация бизнес-логики

Теперь вы можете вносить изменения в код вашего API, используя обновленные поля и новое требование обязательного заполнения номера телефона.

После завершения изменений в коде перезапускайте серверы последовательно один за другим, чтобы уменьшить влияние изменений на пользователей.

Шаг 5: Удаление старых колонок и индексов

Финальная миграция уберёт старые колонки и завершит процесс перехода.

Миграция №3: удаление старого поля full_name.

def upgrade():
    with op.batch_alter_table('users') as batch_op:
        batch_op.drop_column('full_name')

def downgrade():
    op.add_column('users', sa.Column('full_name', sa.String(length=255)))

На данном этапе также рекомендуется удалить временной индекс, созданный ранее (если применялся индекс CONCURRENTLY):

DROP INDEX IF EXISTS idx_users_phone_unique;

Затем создаем новый постоянный уникальный индекс по телефону:

CREATE UNIQUE INDEX idx_users_phone_unique ON users(phone);

Эта операция выполняется через команду CREATE UNIQUE INDEX CONCURRENTLY, чтобы избегать длительных блокировок.

---

Тяжёлые блокировки и как их избежать

В PostgreSQL существуют несколько типов операций, вызывающих долгие блокировки:

• Операции ALTER TABLE без опции WITH NO DATA (например, добавления ограничений типа PRIMARY KEY или FOREIGN KEY).
• Обычные команды CREATE INDEX, DROP INDEX, которые выполняют полное сканирование таблицы.
• Массовые обновления больших таблиц одним большим UPDATE-запросом.

Для минимизации влияния эти операции выполняются поэтапно и с использованием следующих методов:

• Использование конструкции CREATE INDEX CONCURRENTLY: этот способ создает индекс медленно, но асинхронно, без тяжелых блокировок на чтение/запись.
• Масштабирование миграций через небольшие партии с помощью сценария бэч-обновлений.
• Отложенные проверки целостности данных (для массовых изменений полей и условий NOT NULL).

Эти подходы позволяют снизить нагрузку на базу данных и избежать продолжительного снижения производительности или отказа работы приложения.

Проверки после каждой миграции

Перед каждым этапом важно убедиться, что система продолжает стабильно функционировать:

• Проверьте состояние нового индекса и скорость выборок с ним.
• Убедитесь, что новые колонки корректно заполнены данными.
• Выполните нагрузочное тестирование для оценки производительности системы.

Откат на каждом шаге

При возникновении проблем на любом этапе вам потребуется возможность быстрого отката назад.

• Для Step 1: отмена миграции путем удаления временных колонок и временного индекса.
• Для Step 2: восстановить прежние значения телефонов и снять ограничение NOT NULL.
• Для Step 3: вернуть старую колонку обратно.

Каждый шаг требует подготовки обратимых миграций и скриптов отката, которые будут возвращать систему в исходное состояние.

Заключение

Следуя такому плану, вы сможете провести необходимые изменения без длительного простоя сервиса и значительных влияний на производительность системы даже при высоких нагрузках. Важно проводить каждое действие аккуратно, внимательно проверяя промежуточные результаты и стабильность работы сервиса.

Ниже план для PostgreSQL 15 с нулевым простоем и rolling deploy FastAPI-инстансов. Ключевая идея: expand → dual write/dual read → backfill → constraints/index concurrently → contract.

Важно: phone сейчас NULL у ~3% строк. Сделать его NOT NULL и уникальным без бизнес-решения нельзя: нужно либо собрать реальные телефоны, либо удалить/заблокировать такие аккаунты, либо вынести их в отдельный процесс remediation. Генерировать фейковые телефоны обычно плохая идея.

---

1. План по шагам: expand / contract

Фаза 0. Предпроверки на проде

Перед миграциями проверить:

-- Сколько NULL phone
SELECT count(*) 
FROM users 
WHERE phone IS NULL;

-- Дубликаты phone среди NOT NULL
SELECT phone, count(*)
FROM users
WHERE phone IS NOT NULL
GROUP BY phone
HAVING count(*) > 1
ORDER BY count(*) DESC
LIMIT 100;

-- Нестандартные full_name
SELECT full_name
FROM users
WHERE full_name IS NULL OR btrim(full_name) = ''
LIMIT 100;

Если есть дубликаты phone, уникальный индекс не создастся. Их нужно заранее разрулить.

---

Фаза 1. Expand: добавить новые nullable-колонки и синхронизацию

Цель

Добавить:

first_name varchar(255) NULL
last_name varchar(255) NULL

Пока не трогаем full_name.

Для безопасного rolling deploy добавляем trigger, который:

• если старый код пишет только full_name, заполняет first_name / last_name;
• если новый код пишет first_name / last_name, заполняет full_name.

Это защищает от ситуации, когда часть инстансов уже новая, а часть старая.

Деплой 1

1. Alembic migration: добавить колонки и trigger.
2. Приложение пока можно не менять.

---

Фаза 2. Обновить приложение на dual read / dual write

Новый код приложения

При записи пользователя:

• писать first_name;
• писать last_name;
• продолжать писать full_name для совместимости;
• для phone на уровне API запретить NULL для новых и изменяемых записей.

При чтении:

• читать first_name / last_name, если они заполнены;
• fallback на full_name, если backfill ещё не прошёл.

Пример логики:

first_name = user.first_name or parse_first_name(user.full_name)
last_name = user.last_name or parse_last_name(user.full_name)

Деплой

Rolling deploy по одному инстансу за балансировщиком.

После того как все 4 инстанса обновлены, новые записи должны иметь и старые, и новые поля.

---

Фаза 3. Backfill first_name / last_name

Запустить отдельный скрипт, который батчами заполнит first_name / last_name для старых 18 млн строк.

Особенности:

• маленькие батчи;
• commit после каждого батча;
• пауза между батчами;
• FOR UPDATE SKIP LOCKED, чтобы не конфликтовать с приложением;
• лимит по statement timeout / lock timeout.

---

Фаза 4. Подготовить phone

4.1. Запретить новые NULL в приложении

Уже должно быть сделано в фазе 2.

4.2. Разобрать существующие NULL

Для 3% строк нужно бизнес-решение:

Варианты:

1. собрать телефоны у пользователей;
2. заблокировать аккаунты до ввода телефона;
3. удалить/архивировать записи;
4. перенести проблемные записи в отдельное состояние, где users.phone всё равно не NULL.

До устранения NULL нельзя безопасно делать phone NOT NULL.

4.3. Разобрать дубликаты

Уникальный индекс по phone не создастся, если есть дубликаты.

---

Фаза 5. Создать уникальный индекс по phone без блокировки записи

В PostgreSQL использовать:

CREATE UNIQUE INDEX CONCURRENTLY users_phone_uq_idx ON users(phone);

CONCURRENTLY позволяет не блокировать обычные INSERT/UPDATE/DELETE над таблицей надолго.

Важно:

• нельзя выполнять внутри обычной transaction block;
• Alembic migration должна использовать autocommit_block;
• если во время создания появляются дубликаты, индекс может упасть или остаться invalid, его надо удалить и повторить.

---

Фаза 6. Сделать phone NOT NULL

Безопасный путь:

1. Добавить CHECK (phone IS NOT NULL) NOT VALID.
2. Провалидировать constraint.
3. Сделать ALTER COLUMN phone SET NOT NULL.
4. Удалить временный check constraint.

В PostgreSQL 12+ SET NOT NULL может использовать уже валидированный CHECK (phone IS NOT NULL) и избежать полного сканирования таблицы. Но всё равно будет короткий ACCESS EXCLUSIVE lock на изменение метаданных.

Для PostgreSQL 15 этот подход актуален, но поведение и длительность блокировок всё равно нужно проверить на стенде с похожим объёмом.

---

Фаза 7. Contract: убрать full_name

После того как:

• все инстансы приложения используют first_name / last_name;
• backfill завершён;
• мониторинг показывает, что full_name больше не используется;
• триггер больше не нужен.

Делаем:

1. удалить trigger;
2. удалить trigger function;
3. удалить колонку full_name.

DROP COLUMN берёт ACCESS EXCLUSIVE lock, но обычно это быстрая metadata-операция. На нагруженной таблице всё равно лучше выставить lock_timeout, чтобы не повиснуть в очереди.

---

2. Код миграций Alembic

Миграция 001: expand users names

"""001_expand_users_names

Revision ID: 001_expand_users_names
Revises: previous_revision
Create Date: 2026-xx-xx
"""

from alembic import op
import sqlalchemy as sa


revision = "001_expand_users_names"
down_revision = "previous_revision"
branch_labels = None
depends_on = None


def upgrade():
    op.add_column(
        "users",
        sa.Column("first_name", sa.String(length=255), nullable=True),
    )
    op.add_column(
        "users",
        sa.Column("last_name", sa.String(length=255), nullable=True),
    )

    op.execute("""
    CREATE OR REPLACE FUNCTION users_sync_names_trg_fn()
    RETURNS trigger
    LANGUAGE plpgsql
    AS $$
    DECLARE
        normalized_full_name text;
        first_part text;
        rest_part text;
    BEGIN
        /*
          Если новый код передал first_name/last_name, поддерживаем full_name.
          Если старый код передал только full_name, поддерживаем first_name/last_name.
        */

        IF TG_OP = 'INSERT' THEN
            -- New-code path: first_name / last_name -> full_name
            IF NEW.full_name IS NULL
               AND (NEW.first_name IS NOT NULL OR NEW.last_name IS NOT NULL)
            THEN
                NEW.full_name := btrim(concat_ws(' ', NEW.first_name, NEW.last_name));
            END IF;

            -- Old-code path: full_name -> first_name / last_name
            IF NEW.full_name IS NOT NULL
               AND NEW.first_name IS NULL
               AND NEW.last_name IS NULL
            THEN
                normalized_full_name := btrim(regexp_replace(NEW.full_name, '\\s+', ' ', 'g'));
                first_part := split_part(normalized_full_name, ' ', 1);
                rest_part := NULLIF(btrim(substr(
                    normalized_full_name,
                    length(first_part) + 2
                )), '');

                NEW.first_name := NULLIF(first_part, '');
                NEW.last_name := rest_part;
            END IF;

            RETURN NEW;
        END IF;

        IF TG_OP = 'UPDATE' THEN
            -- Если изменили first_name/last_name, обновляем full_name
            IF (NEW.first_name IS DISTINCT FROM OLD.first_name)
               OR (NEW.last_name IS DISTINCT FROM OLD.last_name)
            THEN
                NEW.full_name := btrim(concat_ws(' ', NEW.first_name, NEW.last_name));
                RETURN NEW;
            END IF;

            -- Если изменили full_name старым кодом, обновляем first_name/last_name
            IF NEW.full_name IS DISTINCT FROM OLD.full_name THEN
                normalized_full_name := btrim(regexp_replace(NEW.full_name, '\\s+', ' ', 'g'));
                first_part := split_part(normalized_full_name, ' ', 1);
                rest_part := NULLIF(btrim(substr(
                    normalized_full_name,
                    length(first_part) + 2
                )), '');

                NEW.first_name := NULLIF(first_part, '');
                NEW.last_name := rest_part;
                RETURN NEW;
            END IF;

            RETURN NEW;
        END IF;

        RETURN NEW;
    END;
    $$;
    """)

    op.execute("""
    CREATE TRIGGER users_sync_names_trg
    BEFORE INSERT OR UPDATE OF full_name, first_name, last_name
    ON users
    FOR EACH ROW
    EXECUTE FUNCTION users_sync_names_trg_fn();
    """)


def downgrade():
    op.execute("""
    DROP TRIGGER IF EXISTS users_sync_names_trg ON users;
    """)

    op.execute("""
    DROP FUNCTION IF EXISTS users_sync_names_trg_fn();
    """)

    op.drop_column("users", "last_name")
    op.drop_column("users", "first_name")

---

Миграция 002: создать уникальный индекс по phone concurrently

Перед запуском нужно убедиться, что нет дубликатов phone.

"""002_create_unique_phone_index_concurrently

Revision ID: 002_create_unique_phone_index_concurrently
Revises: 001_expand_users_names
Create Date: 2026-xx-xx
"""

from alembic import op


revision = "002_create_unique_phone_index_concurrently"
down_revision = "001_expand_users_names"
branch_labels = None
depends_on = None


def upgrade():
    conn = op.get_bind()

    # Не даём миграции долго ждать блокировки.
    conn.execute("SET lock_timeout = '5s'")
    conn.execute("SET statement_timeout = '0'")

    # CREATE INDEX CONCURRENTLY нельзя выполнять внутри transaction block.
    with op.get_context().autocommit_block():
        op.execute("""
        CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS users_phone_uq_idx
        ON users (phone);
        """)


def downgrade():
    with op.get_context().autocommit_block():
        op.execute("""
        DROP INDEX CONCURRENTLY IF EXISTS users_phone_uq_idx;
        """)

Комментарий: уникальный индекс в PostgreSQL допускает много NULL, поэтому технически его можно создать до NOT NULL. Но финальная уникальность «каждый пользователь имеет уникальный обязательный phone» появится только после NOT NULL.

---

Миграция 003: добавить и провалидировать check phone is not null

Запускать только после того, как:

SELECT count(*) FROM users WHERE phone IS NULL;

вернул 0.

"""003_validate_phone_not_null_check

Revision ID: 003_validate_phone_not_null_check
Revises: 002_create_unique_phone_index_concurrently
Create Date: 2026-xx-xx
"""

from alembic import op


revision = "003_validate_phone_not_null_check"
down_revision = "002_create_unique_phone_index_concurrently"
branch_labels = None
depends_on = None


def upgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("SET statement_timeout = '0'")

    op.execute("""
    ALTER TABLE users
    ADD CONSTRAINT users_phone_not_null_chk
    CHECK (phone IS NOT NULL)
    NOT VALID;
    """)

    # VALIDATE CONSTRAINT сканирует таблицу, но не блокирует обычные DML как ACCESS EXCLUSIVE.
    op.execute("""
    ALTER TABLE users
    VALIDATE CONSTRAINT users_phone_not_null_chk;
    """)


def downgrade():
    op.execute("""
    ALTER TABLE users
    DROP CONSTRAINT IF EXISTS users_phone_not_null_chk;
    """)

---

Миграция 004: установить phone NOT NULL

"""004_set_phone_not_null

Revision ID: 004_set_phone_not_null
Revises: 003_validate_phone_not_null_check
Create Date: 2026-xx-xx
"""

from alembic import op


revision = "004_set_phone_not_null"
down_revision = "003_validate_phone_not_null_check"
branch_labels = None
depends_on = None


def upgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("SET statement_timeout = '30s'")

    # В PostgreSQL 12+ может использовать валидированный CHECK для избежания полного скана.
    # Но ACCESS EXCLUSIVE lock на изменение метаданных всё равно будет взят кратковременно.
    op.execute("""
    ALTER TABLE users
    ALTER COLUMN phone SET NOT NULL;
    """)

    # Временный CHECK больше не нужен.
    op.execute("""
    ALTER TABLE users
    DROP CONSTRAINT IF EXISTS users_phone_not_null_chk;
    """)


def downgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("""
    ALTER TABLE users
    ALTER COLUMN phone DROP NOT NULL;
    """)

---

Миграция 005: contract, удалить full_name

Запускать только когда:

• все приложения не используют full_name;
• backfill завершён;
• некоторое время прошёл мониторинг.

"""005_drop_full_name

Revision ID: 005_drop_full_name
Revises: 004_set_phone_not_null
Create Date: 2026-xx-xx
"""

from alembic import op
import sqlalchemy as sa


revision = "005_drop_full_name"
down_revision = "004_set_phone_not_null"
branch_labels = None
depends_on = None


def upgrade():
    op.execute("SET lock_timeout = '5s'")
    op.execute("SET statement_timeout = '30s'")

    op.execute("""
    DROP TRIGGER IF EXISTS users_sync_names_trg ON users;
    """)

    op.execute("""
    DROP FUNCTION IF EXISTS users_sync_names_trg_fn();
    """)

    op.drop_column("users", "full_name")


def downgrade():
    op.add_column(
        "users",
        sa.Column("full_name", sa.String(length=255), nullable=True),
    )

    op.execute("""
    UPDATE users
    SET full_name = btrim(concat_ws(' ', first_name, last_name))
    WHERE full_name IS NULL;
    """)

    op.execute("""
    CREATE OR REPLACE FUNCTION users_sync_names_trg_fn()
    RETURNS trigger
    LANGUAGE plpgsql
    AS $$
    BEGIN
        IF (NEW.first_name IS DISTINCT FROM OLD.first_name)
           OR (NEW.last_name IS DISTINCT FROM OLD.last_name)
        THEN
            NEW.full_name := btrim(concat_ws(' ', NEW.first_name, NEW.last_name));
        END IF;

        RETURN NEW;
    END;
    $$;
    """)

    op.execute("""
    CREATE TRIGGER users_sync_names_trg
    BEFORE UPDATE OF first_name, last_name
    ON users
    FOR EACH ROW
    EXECUTE FUNCTION users_sync_names_trg_fn();
    """)

---

3. Скрипт переноса данных батчами

Ниже пример standalone-скрипта на Python. Он:

• берёт по batch_size строк;
• использует FOR UPDATE SKIP LOCKED;
• обновляет только строки, где first_name и last_name ещё не заполнены;
• коммитит каждый батч;
• делает паузу;
• выставляет lock_timeout.

Рекомендация для старта:

• batch_size = 2_000;
• sleep = 0.05–0.2s;
• запускать не в пик;
• наблюдать за latency, locks, replica lag, CPU/IO.

Для 18 млн строк батч 2 000 — это до 9 000 батчей. При 0.1s паузы чисто паузы дадут ~15 минут, плюс время UPDATE. Реально может занять часы — это нормально.

import os
import time
import argparse
import psycopg2
from psycopg2.extras import execute_values


DSN = os.environ["DATABASE_URL"]


def main(batch_size: int, sleep_seconds: float):
    total = 0

    while True:
        with psycopg2.connect(DSN) as conn:
            conn.autocommit = False

            with conn.cursor() as cur:
                cur.execute("SET LOCAL lock_timeout = '500ms'")
                cur.execute("SET LOCAL statement_timeout = '30s'")
                cur.execute("SET LOCAL idle_in_transaction_session_timeout = '30s'")

                cur.execute(
                    """
                    WITH batch AS (
                        SELECT id, full_name
                        FROM users
                        WHERE full_name IS NOT NULL
                          AND (
 first_name IS NULL
 OR last_name IS NULL
                          )
                        ORDER BY id
                        LIMIT %s
                        FOR UPDATE SKIP LOCKED
                    ),
                    parsed AS (
                        SELECT
                            id,
                            NULLIF(split_part(
 btrim(regexp_replace(full_name, '\\s+', ' ', 'g')),
 ' ',
 1
                            ), '') AS parsed_first_name,
                            NULLIF(
 btrim(substr(
 btrim(regexp_replace(full_name, '\\s+', ' ', 'g')),
 length(split_part(
 btrim(regexp_replace(full_name, '\\s+', ' ', 'g')),
 ' ',
 1
 )) + 2
 )),
 ''
                            ) AS parsed_last_name
                        FROM batch
                    )
                    UPDATE users u
                    SET
                        first_name = COALESCE(u.first_name, p.parsed_first_name),
                        last_name = COALESCE(u.last_name, p.parsed_last_name)
                    FROM parsed p
                    WHERE u.id = p.id
                    RETURNING u.id;
                    """,
                    (batch_size,),
                )

                updated = cur.rowcount
                conn.commit()

        total += updated

        if updated == 0:
            print(f"Done. Total updated: {total}")
            break

        print(f"Updated batch: {updated}, total: {total}")

        time.sleep(sleep_seconds)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--batch-size", type=int, default=2000)
    parser.add_argument("--sleep", type=float, default=0.1)
    args = parser.parse_args()

    main(args.batch_size, args.sleep)

Альтернатива для больших объёмов — двигаться по id-диапазонам, если id коррелирует с физическим порядком и есть primary key/index:

UPDATE users
SET ...
WHERE id > :last_id
  AND id <= :last_id + :step
  AND ...

Это часто дешевле, чем каждый раз искать первые незаполненные строки по всей таблице.

---

Скрипт проверки прогресса backfill

SELECT count(*) AS not_backfilled
FROM users
WHERE full_name IS NOT NULL
  AND (first_name IS NULL OR last_name IS NULL);

Можно смотреть по диапазонам:

SELECT
    min(id),
    max(id),
    count(*) FILTER (WHERE first_name IS NULL OR last_name IS NULL) AS remaining
FROM users;

---

Скрипт поиска проблемных phone

-- NULL
SELECT count(*)
FROM users
WHERE phone IS NULL;

-- Дубликаты
SELECT phone, count(*)
FROM users
WHERE phone IS NOT NULL
GROUP BY phone
HAVING count(*) > 1
ORDER BY count(*) DESC
LIMIT 100;

---

4. Какие операции берут тяжёлые блокировки и как их избежать

4.1. ALTER TABLE ... ADD COLUMN

В PostgreSQL берёт ACCESS EXCLUSIVE lock, но если колонка nullable и без volatile default, операция быстрая metadata-only.

Безопасно:

ALTER TABLE users ADD COLUMN first_name varchar(255);

Опаснее:

ALTER TABLE users ADD COLUMN some_col text NOT NULL DEFAULT expensive_function();

В PostgreSQL 11+ добавление колонки с constant default стало metadata-only, но всё равно ACCESS EXCLUSIVE lock кратковременно берётся. Для нагруженной базы лучше:

1. добавить nullable колонку без default;
2. backfill батчами;
3. поставить default отдельно, если нужен;
4. потом constraint.

---

4.2. ALTER TABLE ... DROP COLUMN

Берёт ACCESS EXCLUSIVE lock.

Обычно быстро, но на 1200 RPS может встать в очередь за активными транзакциями и заблокировать последующие запросы.

Как снижать риск:

SET lock_timeout = '5s';
ALTER TABLE users DROP COLUMN full_name;

Если не получил lock быстро — миграция падает, повторить позже.

---

4.3. ALTER TABLE ... ALTER COLUMN ... SET NOT NULL

Берёт ACCESS EXCLUSIVE lock.

Без подготовки может сканировать всю таблицу, что плохо для 18 млн строк.

Безопаснее:

ALTER TABLE users
ADD CONSTRAINT users_phone_not_null_chk
CHECK (phone IS NOT NULL)
NOT VALID;

ALTER TABLE users
VALIDATE CONSTRAINT users_phone_not_null_chk;

ALTER TABLE users
ALTER COLUMN phone SET NOT NULL;

В PostgreSQL 12+ SET NOT NULL может использовать валидированный check constraint, чтобы избежать полного table scan. В PostgreSQL 15 это должно работать, но всё равно нужно проверить на стенде.

---

4.4. CREATE INDEX

Обычный:

CREATE UNIQUE INDEX users_phone_uq_idx ON users(phone);

Блокирует записи в таблицу на время построения индекса. Для 18 млн строк это неприемлемо.

Нужно:

CREATE UNIQUE INDEX CONCURRENTLY users_phone_uq_idx ON users(phone);

Особенности:

• нельзя внутри transaction block;
• работает дольше обычного;
• требует больше внимания к maintenance work mem / IO;
• может оставить invalid index при ошибке;
• всё равно берёт краткие locks в начале/конце.

Проверка invalid index:

SELECT
    c.relname,
    i.indisvalid,
    i.indisready
FROM pg_index i
JOIN pg_class c ON c.oid = i.indexrelid
WHERE c.relname = 'users_phone_uq_idx';

Удаление invalid index:

DROP INDEX CONCURRENTLY IF EXISTS users_phone_uq_idx;

---

4.5. ALTER TABLE ... ADD CONSTRAINT UNIQUE

Если делать напрямую, может взять тяжёлые блокировки и строить индекс не concurrent.

Лучше:

1. CREATE UNIQUE INDEX CONCURRENTLY;
2. при необходимости прикрепить constraint:

ALTER TABLE users
ADD CONSTRAINT users_phone_uq
UNIQUE USING INDEX users_phone_uq_idx;

В этом кейсе можно оставить просто unique index, если приложению/ORM не нужна именно named constraint.

---

4.6. Длинные транзакции

Даже лёгкая DDL-операция может застрять, если есть длинная транзакция, держащая lock.

Перед DDL проверять:

SELECT
    pid,
    usename,
    state,
    now() - xact_start AS xact_age,
    query
FROM pg_stat_activity
WHERE xact_start IS NOT NULL
ORDER BY xact_start;

И блокировки:

SELECT
    blocked.pid AS blocked_pid,
    blocked.query AS blocked_query,
    blocking.pid AS blocking_pid,
    blocking.query AS blocking_query
FROM pg_locks blocked_locks
JOIN pg_stat_activity blocked
    ON blocked.pid = blocked_locks.pid
JOIN pg_locks blocking_locks
    ON blocking_locks.locktype = blocked_locks.locktype
   AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
   AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
   AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page
   AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple
   AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid
   AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid
   AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid
   AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid
   AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid
   AND blocking_locks.pid != blocked_locks.pid
JOIN pg_stat_activity blocking
    ON blocking.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

---

5. План отката и проверки после каждого шага

Шаг 001: add first_name / last_name / trigger

Откат

Безопасно откатить, пока новый код не зависит от колонок:

alembic downgrade previous_revision

Или руками:

DROP TRIGGER IF EXISTS users_sync_names_trg ON users;
DROP FUNCTION IF EXISTS users_sync_names_trg_fn();
ALTER TABLE users DROP COLUMN IF EXISTS first_name;
ALTER TABLE users DROP COLUMN IF EXISTS last_name;

Проверить

\d users

SELECT column_name, is_nullable
FROM information_schema.columns
WHERE table_name = 'users'
  AND column_name IN ('full_name', 'first_name', 'last_name');

Проверить trigger:

INSERT INTO users (id, full_name, phone, created_at)
VALUES (-1, 'Ivan Petrov', '+70000000001', now());

SELECT full_name, first_name, last_name
FROM users
WHERE id = -1;

DELETE FROM users WHERE id = -1;

---

Шаг 002: deploy dual read / dual write

Откат

Откатить код на старую версию. Благодаря trigger старый код продолжит работать через full_name.

Проверить

Создать/обновить пользователя через API и проверить:

SELECT full_name, first_name, last_name, phone
FROM users
WHERE id = :id;

Также проверить, что API больше не создаёт phone = NULL.

SELECT count(*)
FROM users
WHERE phone IS NULL
  AND created_at > now() - interval '1 hour';

---

Шаг 003: backfill names

Откат

Обычно откатывать backfill не нужно: данные добавлены в новые колонки, старый full_name не тронут.

Если нужно очистить:

UPDATE users
SET first_name = NULL,
    last_name = NULL;

Но это тяжёлая операция на 18 млн строк, делать только при крайней необходимости и тоже батчами.

Проверить

SELECT count(*)
FROM users
WHERE full_name IS NOT NULL
  AND first_name IS NULL;

SELECT id, full_name, first_name, last_name
FROM users
WHERE full_name IS NOT NULL
  AND (first_name IS NULL OR last_name IS NULL)
LIMIT 100;

---

Шаг 004: remediation phone NULL / duplicates

Откат

Зависит от бизнес-процесса:

• если аккаунты блокировались — разблокировать;
• если phone обновлялся — вернуть из audit/history;
• если записи удалялись — восстановить из backup/archive.

Перед массовыми изменениями желательно сохранить affected rows:

CREATE TABLE users_phone_remediation_backup AS
SELECT id, phone
FROM users
WHERE phone IS NULL
   OR phone IN (
        SELECT phone
        FROM users
        WHERE phone IS NOT NULL
        GROUP BY phone
        HAVING count(*) > 1
   );

Проверить

SELECT count(*)
FROM users
WHERE phone IS NULL;

SELECT phone, count(*)
FROM users
WHERE phone IS NOT NULL
GROUP BY phone
HAVING count(*) > 1
LIMIT 10;

Оба запроса должны вернуть отсутствие проблем.

---

Шаг 005: create unique index concurrently

Откат

DROP INDEX CONCURRENTLY IF EXISTS users_phone_uq_idx;

Если индекс invalid:

DROP INDEX CONCURRENTLY IF EXISTS users_phone_uq_idx;

Затем исправить данные и повторить.

Проверить

SELECT
    c.relname,
    i.indisunique,
    i.indisvalid,
    i.indisready
FROM pg_index i
JOIN pg_class c ON c.oid = i.indexrelid
WHERE c.relname = 'users_phone_uq_idx';

Проверить, что дубль не вставляется:

INSERT INTO users (id, full_name, first_name, last_name, phone, created_at)
VALUES (-1, 'Test User', 'Test', 'User', '+79999999999', now());

INSERT INTO users (id, full_name, first_name, last_name, phone, created_at)
VALUES (-2, 'Test User2', 'Test', 'User2', '+79999999999', now());
-- должно упасть

DELETE FROM users WHERE id IN (-1, -2);

---

Шаг 006: phone NOT NULL

Откат

ALTER TABLE users
ALTER COLUMN phone DROP NOT NULL;

Если временный check ещё есть:

ALTER TABLE users
DROP CONSTRAINT IF EXISTS users_phone_not_null_chk;

Проверить

SELECT is_nullable
FROM information_schema.columns
WHERE table_name = 'users'
  AND column_name = 'phone';

Должно быть:

NO

Проверить вставку:

INSERT INTO users (id, first_name, last_name, phone, created_at)
VALUES (-1, 'Test', 'User', NULL, now());
-- должно упасть

---

Шаг 007: contract drop full_name

Откат

Можно вернуть колонку:

ALTER TABLE users ADD COLUMN full_name varchar(255);

UPDATE users
SET full_name = btrim(concat_ws(' ', first_name, last_name))
WHERE full_name IS NULL;

Но UPDATE на 18 млн строк тяжёлый, делать батчами. Поэтому перед удалением full_name лучше выдержать период наблюдения, например 1–2 релиза.

Проверить перед drop

На уровне приложения:

• нет обращений к full_name в коде;
• нет ошибок в логах;
• нет старых инстансов;
• отчёты на реплике не читают full_name.

На уровне БД можно временно включить логирование запросов или посмотреть pg_stat_statements, если установлен:

SELECT query, calls
FROM pg_stat_statements
WHERE query ILIKE '%full_name%'
ORDER BY calls DESC
LIMIT 20;

Проверить после drop

SELECT column_name
FROM information_schema.columns
WHERE table_name = 'users'
ORDER BY ordinal_position;

---

Дополнительные практические рекомендации

Запуск DDL с timeout

Для всех DDL на горячей таблице:

SET lock_timeout = '5s';
SET statement_timeout = '30s';

Для долгого CREATE INDEX CONCURRENTLY:

SET lock_timeout = '5s';
SET statement_timeout = '0';

Если DDL не получил lock быстро — лучше упасть и повторить позже, чем заблокировать очередь запросов.

---

Мониторинг во время миграции

Следить за:

• latency API;
• ошибками 5xx;
• PostgreSQL CPU / IO;
• pg_stat_activity;
• количеством locks;
• replication lag;
• размером WAL;
• autovacuum;
• bloat таблицы после массового backfill.

После backfill может вырасти bloat. Не делать VACUUM FULL, он берёт тяжёлую блокировку. Использовать обычный VACUUM, autovacuum tuning или pg_repack, если он доступен и протестирован.

---

Обязательно проверить на стенде

Поскольку блокировки и длительность зависят от:

• версии PostgreSQL;
• размера таблицы;
• наличия длинных транзакций;
• индексов;
• FK;
• активности autovacuum;
• параметров maintenance_work_mem;
• железа и IO,

нужно прогнать на стенде с копией схемы и близким объёмом данных:

1. ADD COLUMN;
2. trigger overhead на write path;
3. backfill с разными batch size;
4. CREATE UNIQUE INDEX CONCURRENTLY;
5. VALIDATE CONSTRAINT;
6. ALTER COLUMN SET NOT NULL;
7. DROP COLUMN.

Для PostgreSQL 15 описанный план является штатным zero-downtime-подходом, но финальные тайминги и lock-поведение всё равно нужно подтвердить нагрузочным тестом.

План миграции users без даунтайма (PostgreSQL 15 / Alembic / FastAPI, 4 инстанса)

0. Предположения и то, что нужно зафиксировать до старта (не техническая, а организационная часть)

1. Правило разбиения full_name — нужно решить с продактом: первое слово = first_name, остаток = last_name (либо наоборот), как быть с одним словом, двойными пробелами, NULL. Беру за основу: первое слово → first_name, всё остальное → last_name, если слово одно — last_name = '' (или NULL, зависит от требований).
2. 3% NULL по phone — это не чисто техническая проблема: нельзя honestly поставить NOT NULL, если у 540k пользователей физически нет номера. Варианты:
3. бизнес-процесс дособора номеров (support/CRM) с дедлайном;
4. временная пометка аккаунтов флагом phone_required_action = true и отсрочка контракта;
5. в крайнем случае — синтетический placeholder (не рекомендую, ломает семантику поля и UNIQUE).

Я закладываю в план отдельный флаг и возможность растянуть constraint по времени, не блокируя остальную миграцию.

1. Проверить на реплику — она физическая (streaming), DDL реплицируется автоматически, отдельных миграций на ней делать не нужно, но replication lag нужно мониторить во время батчей (большой объём UPDATE/WAL).
2. Весь план прогнать на стейджинге с дампом прод-данных и PostgreSQL той же минорной версии — поведение блокировок (SET NOT NULL, ADD COLUMN DEFAULT) менялось между 11→12→13, и лучше подтвердить именно на PG15.

---

1. Стратегия: Expand → Migrate → Contract, два независимых трека

Трек A — full_name → first_name/last_name:
Expand (новые колонки) → dual-write в коде → backfill батчами → переключение чтения → Contract (drop full_name).

Трек B — phone NOT NULL + UNIQUE:
Partial unique index (не ждём 100% заполненности) → backfill недостающих номеров (бизнес) → CHECK ... NOT VALID → VALIDATE CONSTRAINT → SET NOT NULL → финальная чистка индекса.

Оба трека используют только:

• ADD COLUMN nullable (мгновенно, метаданные)
• CREATE INDEX CONCURRENTLY
• ADD CONSTRAINT ... NOT VALID + VALIDATE CONSTRAINT
• rolling-деплой приложения по 1 инстансу, при этом на каждом шаге обе версии кода (N и N+1) должны уметь работать с одной и той же схемой, т.к. 4 инстанса какое-то время будут разных версий.

---

2. Пошаговый план (деплои пронумерованы по порядку)

Трек A: full_name

Трек B: phone

Треки можно вести параллельно, они не блокируют друг друга.

---

3. Код миграций Alembic

A1 — добавить колонки

"""A1: add first_name/last_name

Revision ID: a1_add_name_columns
"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users', sa.Column('first_name', sa.String(255), nullable=True))
    op.add_column('users', sa.Column('last_name', sa.String(255), nullable=True))

def downgrade():
    op.drop_column('users', 'last_name')
    op.drop_column('users', 'first_name')

Это чистая операция с метаданными каталога — ACCESS EXCLUSIVE lock берётся, но держится миллисекунды (нет дефолта, требующего переписывания таблицы).

A7 — удалить full_name (выполнять только после недель наблюдения)

"""A7: drop full_name

Revision ID: a7_drop_full_name
"""
from alembic import op

def upgrade():
    op.execute("SET lock_timeout = '2s'")
    op.drop_column('users', 'full_name')

def downgrade():
    # Данные физически потеряны, этот downgrade не восстанавливает их,
    # только возвращает колонку. Реальное восстановление — из бэкапа/PITR.
    op.add_column('users', sa.Column('full_name', sa.String(255), nullable=True))

B2 — partial unique index (CONCURRENTLY, вне транзакции)

"""B2: partial unique index on phone

Revision ID: b2_phone_partial_unique_index
"""
from alembic import op

def upgrade():
    with op.get_context().autocommit_block():
        op.execute("""
            CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ux_users_phone_partial
            ON users (phone)
            WHERE phone IS NOT NULL
        """)

def downgrade():
    with op.get_context().autocommit_block():
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ux_users_phone_partial")

Важно: если CREATE INDEX CONCURRENTLY упадёт (например, обнаружены дубликаты), индекс остаётся в состоянии INVALID — его нужно дропнуть (DROP INDEX CONCURRENTLY) и перезапустить после устранения дублей. Автоматический retry в коде миграции делать не стоит — обязательно разобраться с причиной.

B3 — флаг для неполных данных

"""B3: phone_required_action flag"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column(
        'users',
        sa.Column('phone_required_action', sa.Boolean(), nullable=False, server_default=sa.false())
    )
    # server_default — константа, в PG11+ не переписывает таблицу, быстро

def downgrade():
    op.drop_column('users', 'phone_required_action')

B5/B6 — CHECK NOT VALID + VALIDATE

"""B5-B6: not-null check for phone"""
from alembic import op

def upgrade():
    op.execute("SET lock_timeout = '2s'")
    op.execute("""
        ALTER TABLE users
        ADD CONSTRAINT chk_users_phone_not_null
        CHECK (phone IS NOT NULL) NOT VALID
    """)
    # Валидация — отдельный шаг, вне общей транзакции с ADD CONSTRAINT,
    # чтобы не держать лишний лок во время долгого скана
    with op.get_context().autocommit_block():
        op.execute("ALTER TABLE users VALIDATE CONSTRAINT chk_users_phone_not_null")

def downgrade():
    op.execute("ALTER TABLE users DROP CONSTRAINT IF EXISTS chk_users_phone_not_null")

VALIDATE CONSTRAINT берёт SHARE UPDATE EXCLUSIVE — конфликтует только с другим DDL, не блокирует SELECT/INSERT/UPDATE/DELETE. Делает полный скан таблицы, но в фоне, это I/O-нагрузка — желательно запускать не в пик 1200 rps, а в окно пониженной нагрузки (хоть явного maintenance window нет, всё равно есть суточная цикличность).

B7 — финальный NOT NULL

"""B7: set phone not null"""
from alembic import op

def upgrade():
    op.execute("SET lock_timeout = '2s'")
    # В PG12+ при наличии валидного CHECK (phone IS NOT NULL) планировщик
    # пропускает повторный полный скан — проверить на стенде именно на PG15.
    op.execute("ALTER TABLE users ALTER COLUMN phone SET NOT NULL")
    op.execute("ALTER TABLE users DROP CONSTRAINT IF EXISTS chk_users_phone_not_null")

def downgrade():
    op.execute("ALTER TABLE users ALTER COLUMN phone DROP NOT NULL")

B8 — финальная чистка индекса (опционально)

"""B8: replace partial unique index with full unique constraint"""
from alembic import op

def upgrade():
    with op.get_context().autocommit_block():
        op.execute("""
            CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ux_users_phone
            ON users (phone)
        """)
    op.execute("""
        ALTER TABLE users
        ADD CONSTRAINT ux_users_phone UNIQUE USING INDEX ux_users_phone
    """)  # присоединение готового индекса как constraint — только метаданные
    with op.get_context().autocommit_block():
        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ux_users_phone_partial")

def downgrade():
    op.execute("ALTER TABLE users DROP CONSTRAINT IF EXISTS ux_users_phone")
    with op.get_context().autocommit_block():
        op.execute("""
            CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ux_users_phone_partial
            ON users (phone) WHERE phone IS NOT NULL
        """)

Замечание: PostgreSQL не позволяет превратить partial index напрямую в table constraint (ADD CONSTRAINT ... UNIQUE USING INDEX требует непарциального индекса), поэтому здесь создаётся новый полный индекс.

---

4. Скрипт батч-переноса данных (backfill first_name/last_name)

Выполняется отдельным процессом (не Alembic-транзакцией), вне долгой транзакции, идемпотентно, с keyset-пагинацией по id (не OFFSET — на 18 млн строк это дорого), с throttling по replication lag.

import time
import logging
import psycopg2
import psycopg2.extras

log = logging.getLogger("backfill")

BATCH_SIZE = 5000
SLEEP_BETWEEN_BATCHES = 0.15          # сек, базовая пауза
MAX_REPLICATION_LAG_BYTES = 50 * 1024 * 1024  # 50 MB — порог для троттлинга
DSN = "postgresql://user:pass@primary-host/db"

def get_replication_lag(conn):
    with conn.cursor() as cur:
        cur.execute("""
            SELECT COALESCE(MAX(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)), 0)
            FROM pg_stat_replication
        """)
        return cur.fetchone()[0]

def split_full_name(full_name: str):
    if not full_name:
        return None, None
    parts = full_name.strip().split(maxsplit=1)
    if len(parts) == 1:
        return parts[0], ""
    return parts[0], parts[1]

def run():
    conn = psycopg2.connect(DSN)
    conn.autocommit = False

    last_id = 0
    total = 0

    with conn.cursor() as cur:
        cur.execute("SELECT max(id) FROM users")
        max_id = cur.fetchone()[0] or 0

    while last_id < max_id:
        lag = get_replication_lag(conn)
        if lag > MAX_REPLICATION_LAG_BYTES:
            log.warning("replication lag %.1fMB, pausing 5s", lag / 1024 / 1024)
            time.sleep(5)
            continue

        with conn.cursor() as cur:
            cur.execute("SET lock_timeout = '2s'")
            cur.execute("SET statement_timeout = '5s'")

            # идемпотентно: обрабатываем только то, что ещё не заполнено
            cur.execute("""
                SELECT id, full_name
                FROM users
                WHERE id > %s AND id <= %s
                  AND full_name IS NOT NULL
                  AND first_name IS NULL
                ORDER BY id
                LIMIT %s
            """, (last_id, last_id + BATCH_SIZE * 10, BATCH_SIZE))
            rows = cur.fetchall()

        if not rows:
            last_id += BATCH_SIZE * 10
            continue

        updates = []
        for row_id, full_name in rows:
            first, last = split_full_name(full_name)
            updates.append((first, last, row_id))

        try:
            with conn.cursor() as cur:
                psycopg2.extras.execute_batch(cur, """
                    UPDATE users
                    SET first_name = %s, last_name = %s
                    WHERE id = %s AND first_name IS NULL
                """, updates, page_size=500)
            conn.commit()
        except Exception:
            conn.rollback()
            log.exception("batch failed at id=%s, retrying after pause", last_id)
            time.sleep(2)
            continue

        last_id = rows[-1][0]
        total += len(rows)
        log.info("processed up to id=%s, total=%s", last_id, total)
        time.sleep(SLEEP_BETWEEN_BATCHES)

    log.info("backfill finished, total rows=%s", total)

if __name__ == "__main__":
    run()

Ключевые моменты:

• батч 5000 строк — компромисс между числом транзакций и временем удержания лока в рамках одной транзакции (UPDATE на 5000 строк по PK держит row-level locks, не table-level, и быстро коммитится);
• условие WHERE first_name IS NULL в самом UPDATE — защита от гонки, если скрипт запущен повторно или параллельно с dual-write из приложения;
• пауза 150 мс между батчами — снижает нагрузку на I/O и WAL, не даёт утилизировать диск на 100% в пик;
• троттлинг по pg_stat_replication — чтобы не увеличивать лаг на реплике для отчётов;
• lock_timeout/statement_timeout — при конкуренции скрипт быстро фейлится и ретраит, а не встаёт в очередь локов надолго.

Для трека B (phone_required_action) отдельный backfill не нужен — это внешний бизнес-процесс (CRM/support), триггерящий обычные UPDATE users SET phone = ..., phone_required_action = false WHERE id = ... через штатное приложение.

---

5. Какие операции берут тяжёлые блокировки и как их избегать

Важно про версионность: оптимизация «SET NOT NULL пропускает скан при наличии валидного CHECK» появилась в PostgreSQL 12. Поведение ADD COLUMN DEFAULT без переписывания — PostgreSQL 11+. Эти детали нужно подтвердить прогоном EXPLAIN (ANALYZE, BUFFERS) / замером времени на стенде с PG15 и объёмом, близким к проду (18 млн строк), т.к. в анонсах фич иногда остаются нюансы (например, поведение с expression/volatile constraints отличается).

---

6. Откат по шагам и post-check

A1 (добавление колонок)

• Откат: DROP COLUMN first_name/last_name — безопасно, данных ещё нет в проде-коде.
• Проверить: \d users в psql, alembic current совпадает, нет роста латентности при обычных запросах (колонка nullable без дефолта не влияет на существующие планы).

A2 (dual write в коде)

• Откат: откат деплоя на предыдущий образ — безопасно, т.к. колонки nullable, старый код их просто не трогает.
• Проверить: метрики приложения (error rate, latency), логи на ошибки сериализации (Pydantic-модель должна допускать None в новых полях).

A3 (backfill)

• Откат: не требуется откатывать схему; скрипт идемпотентен, можно остановить (Ctrl+C/SIGTERM) и перезапустить с того же места (условие first_name IS NULL само найдёт недоделанное).
• Проверить: SELECT count(*) FROM users WHERE full_name IS NOT NULL AND first_name IS NULL → должно стремиться к 0; replication lag в норме в течение всего прогона.

A4/A5 (переключение чтения/записи)

• Откат: откат деплоя образа. Безопасно, пока full_name ещё физически существует и заполнена.
• Проверить: сравнить случайную выборку full_name vs first_name || ' ' || last_name на корректность разбиения перед тем как выключать запись в full_name.

A7 (drop full_name)

• Откат: технически можно вернуть колонку (ADD COLUMN full_name), но данные не восстановятся — нужен PITR/бэкап. Поэтому:
• перед этим шагом сделать pg_dump только этой таблицы или снапшот,
• выдержать паузу наблюдения 1-2 недели после A5,
• убедиться, что ни один код (включая batch-джобы, аналитику на реплике, экспорт в DWH) не ссылается на full_name.
• Проверить после: размер таблицы (bloat), что VACUUM отработал, что ни одна джоба не упала с column "full_name" does not exist.

B2 (partial unique index)

• Откат: DROP INDEX CONCURRENTLY — не блокирует.
• Проверить: индекс не в состоянии INVALID (SELECT indisvalid FROM pg_index WHERE indexrelid = 'ux_users_phone_partial'::regclass), нет роста latency на INSERT/UPDATE users (индекс пишется на каждый апдейт телефона).

B3 (флаг)

• Откат: DROP COLUMN phone_required_action — безопасно, если он ещё не используется в бизнес-логике блокировки доступа.
• Проверить: корректность проставления флага на выборке.

B5/B6 (CHECK NOT VALID + VALIDATE)

• Откат: DROP CONSTRAINT chk_users_phone_not_null — мгновенно и безопасно в любой момент.
• Проверить после VALIDATE: constraint convalidated = true в pg_constraint; отсутствие ошибок check constraint violated в логах во время валидации (если есть — значит backfill B4 ещё не завершён, нужно откатить constraint и продолжить сбор данных).

B7 (SET NOT NULL)

• Откат: ALTER COLUMN phone DROP NOT NULL — мгновенно, безопасно.
• Проверить: attnotnull в pg_attribute = true; приложение корректно обрабатывает 23502 (not_null_violation) на случай гонок в коде, которые раньше проходили с NULL.

B8 (финальный unique constraint)

• Откат: вернуть partial index (код в downgrade выше).
• Проверить: pg_constraint содержит ux_users_phone как UNIQUE, старый partial индекс удалён, размер индекса в норме, нет дублирующих индексов, съедающих место и замедляющих запись.

Общие post-deploy проверки на каждом шаге

• pg_stat_activity / pg_locks — нет затянувшихся ACCESS EXCLUSIVE ожиданий;
• replication lag на отчётной реплике;
• p95/p99 latency и error rate (5xx) в APM по users-эндпоинтам;
• ANALYZE users; после массовых batch-апдейтов, чтобы планировщик обновил статистику;
• на стенде перед каждым прод-прогоном — то же самое с объёмом данных, сопоставимым с продом (18 млн строк), чтобы оценить реальное время VALIDATE CONSTRAINT и CREATE INDEX CONCURRENTLY.

Отлично, задача интересная и требует тщательного планирования. Как backend-разработчик с опытом миграций на нагруженных базах, я понимаю критичность отсутствия простоя и необходимость минимизации рисков.

СУБД: PostgreSQL 15
Инструмент миграций: Alembic
Приложение: FastAPI, 4 инстанса за балансировщиком
Текущая схема: users (18 млн строк): id bigint, full_name varchar(255), phone varchar(20) NULL, created_at timestamptz
Изменения:

1. Разделить full_name на first_name и last_name.
2. Сделать phone обязательным (сейчас 3% строк с NULL).
3. Добавить уникальный индекс по phone.

Нагрузка: пик 1 200 RPS
Ограничения: нет окна обслуживания, деплой по одному инстансу, есть реплика для отчётов.

---

Общий подход: Expand and Contract

Мы будем использовать подход "Expand and Contract" для изменения схемы, чтобы обеспечить нулевой простой. Это означает, что мы сначала "расширяем" схему, добавляя новые поля и функциональность, затем переносим данные, а потом "сжимаем" схему, удаляя старые поля.

Важные соображения по PostgreSQL и блокировкам

• ALTER TABLE ADD COLUMN: В PostgreSQL 11+ добавление столбца с DEFAULT значением, которое не является NULL, или с NOT NULL ограничением, является неблокирующей операцией, если DEFAULT значение не вычисляется для каждой строки. Если DEFAULT значение статично, PostgreSQL просто обновляет метаданные таблицы. Однако, если мы добавляем NOT NULL столбец без DEFAULT значения, или с DEFAULT значением, которое нужно вычислить, это может потребовать перезаписи таблицы и блокировки.
• ALTER TABLE ADD CONSTRAINT NOT NULL: Эта операция требует сканирования всей таблицы для проверки существующего данных. Если таблица большая, это может занять много времени и блокировать запись. Мы будем использовать двухшаговый подход: сначала ADD CONSTRAINT ... NOT VALID, затем VALIDATE CONSTRAINT.
• CREATE UNIQUE INDEX CONCURRENTLY: Это критически важная операция для больших таблиц. CONCURRENTLY позволяет создавать индекс без блокировки записи в таблицу, но может быть медленнее и требует двух сканирований таблицы. Если CONCURRENTLY не используется, CREATE UNIQUE INDEX блокирует запись.
• ALTER TABLE DROP COLUMN: Эта операция обычно быстрая, так как просто удаляет метаданные, но может потребовать очистки данных на диске позже.

План по шагам (Expand и Contract)

Фаза 1: Expand (Добавляем новые поля и функциональность)

Цель: Добавить first_name, last_name, подготовить phone к NOT NULL и уникальному индексу, не нарушая работу приложения.

Шаг 1.1: Добавление новых столбцов first_name, last_name (nullable) и временного phone_new (nullable).

• Миграция Alembic (upgrade):
• Добавить first_name (varchar(255), nullable=True).
• Добавить last_name (varchar(255), nullable=True).
• Добавить phone_new (varchar(20), nullable=True) – это будет временный столбец для нового phone с уникальным индексом.
• Создать уникальный индекс на phone_new CONCURRENTLY.
• Изменения в коде приложения (API):
• Модели ORM: Добавить first_name, last_name, phone_new как необязательные поля.
• Логика записи: При создании/обновлении пользователя, если full_name предоставлен, парсить его на first_name и last_name. Если phone предоставлен, записывать его в phone_new.
• Логика чтения: При чтении, если first_name и last_name существуют, использовать их. Иначе, использовать full_name. При чтении phone, отдавать phone_new если он есть, иначе phone.
• Деплой:
• Применить миграцию Alembic на базу.
• Задеплоить новый код на один инстанс FastAPI. Проверить логи, метрики.
• Постепенно задеплоить на остальные инстансы.
• Проверка после:
• Приложение работает без ошибок.
• Новые столбцы появились в схеме базы данных.
• Уникальный индекс на phone_new создан и валиден.
• Новые пользователи могут быть созданы с first_name, last_name и phone_new.

Шаг 1.2: Перенос данных из full_name в first_name и last_name, и из phone в phone_new.

• Скрипт переноса данных:
• Запускается как отдельный процесс или через Celery/RQ.
• Итерирует по таблице users батчами.
• Для каждого пользователя:
• Парсит full_name на first_name и last_name.
• Копирует phone в phone_new.
• Важно: Для строк, где phone сейчас NULL (3%), нужно либо пропустить их, либо присвоить временное уникальное значение в phone_new (например, NULL или temp_phone_<id>). Мы будем присваивать NULL на этом шаге, так как phone_new пока nullable.
• Обновляет строку в базе данных.
• Размер батча: 10 000 - 50 000 строк (зависит от ширины строк и IOPS дисков).
• Паузы: 100 - 500 мс между батчами, чтобы снизить нагрузку на базу.
• Используем UPDATE ... WHERE id IN (...) или UPDATE ... FROM (SELECT id, ... FROM users WHERE ... ORDER BY id LIMIT ...) AS subquery WHERE users.id = subquery.id; для эффективности.
• Проверка после:
• Большинство существующих пользователей имеют заполненные first_name, last_name и phone_new (если phone был не NULL).
• Приложение продолжает работать без проблем.

Шаг 1.3: Обработка NULL значений в phone и phone_new.

• Цель: Сделать phone обязательным. Сейчас 3% строк с NULL.
• Скрипт переноса данных (дополнительный проход):
• Идентифицировать пользователей, у которых phone (и, соответственно, phone_new) равен NULL.
• Для этих пользователей, либо:
• Вариант A (предпочтительный): Связаться с пользователями для получения номера телефона. Это может быть длительный процесс.
• Вариант B (технический): Присвоить временное, уникальное, но невалидное значение (например, temp_phone_missing_<id>) в phone_new. Это позволит нам добавить NOT NULL ограничение. Позже эти пользователи должны будут обновить свой номер телефона.
• Вариант C (если бизнес-логика позволяет): Удалить этих пользователей (маловероятно).
• Предположим, мы выбираем Вариант B. Скрипт обновляет phone_new для всех NULL значений.
• Проверка после:
• Все строки в phone_new имеют не NULL значения.
• Уникальный индекс на phone_new остается валидным.

Шаг 1.4: Добавление NOT NULL ограничения на phone_new.

• Миграция Alembic (upgrade):
• ALTER TABLE users ALTER COLUMN phone_new SET NOT NULL; (Эта операция быстрая, так как мы уже убедились, что нет NULL значений).
• ALTER TABLE users ADD CONSTRAINT unique_phone_new UNIQUE (phone_new) NOT VALID; (Добавляем уникальный индекс как ограничение, но не валидируем его сразу, так как он уже создан CONCURRENTLY).
• ALTER TABLE users VALIDATE CONSTRAINT unique_phone_new; (Валидация ограничения, которая должна быть быстрой, так как индекс уже существует и валиден).
• Изменения в коде приложения (API):
• Модели ORM: Обновить phone_new как обязательное поле.
• Логика записи: Убедиться, что phone_new всегда предоставляется при создании/обновлении.
• Деплой:
• Применить миграцию Alembic на базу.
• Задеплоить новый код на один инстанс FastAPI. Проверить логи, метрики.
• Постепенно задеплоить на остальные инстансы.
• Проверка после:
• Приложение работает без ошибок.
• phone_new теперь NOT NULL в схеме.
• Новое ограничение unique_phone_new активно и валидно.

Фаза 2: Contract (Удаляем старые поля)

Цель: Переключить приложение на использование новых полей, удалить старые.

Шаг 2.1: Переименование столбцов и удаление старого phone.

• Миграция Alembic (upgrade):
• ALTER TABLE users RENAME COLUMN phone_new TO phone; (Переименовываем phone_new в phone).
• ALTER TABLE users DROP COLUMN phone; (Удаляем старый phone столбец).
• ALTER TABLE users DROP COLUMN full_name; (Удаляем full_name).
• Изменения в коде приложения (API):
• Модели ORM: Удалить full_name, phone_new. Обновить phone как обязательное поле.
• Логика записи: Использовать только first_name, last_name, phone.
• Логика чтения: Использовать только first_name, last_name, phone.
• Деплой:
• Применить миграцию Alembic на базу.
• Задеплоить новый код на один инстанс FastAPI. Проверить логи, метрики.
• Постепенно задеплоить на остальные инстансы.
• Проверка после:
• Приложение работает без ошибок.
• Старые столбцы full_name и phone удалены.
• Новые столбцы first_name, last_name и обновленный phone используются.
• Уникальный индекс на phone активен.

---

2. Код миграций для каждого шага (Alembic)

Предполагается, что Alembic уже настроен.

Шаг 1.1: Добавление новых столбцов и уникального индекса

# alembic/versions/xxxxxxxxxxxx_add_first_last_name_and_phone_new.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Добавляем first_name и last_name как nullable
    op.add_column('users', sa.Column('first_name', sa.String(length=255), nullable=True))
    op.add_column('users', sa.Column('last_name', sa.String(length=255), nullable=True))

    # Добавляем phone_new как nullable
    op.add_column('users', sa.Column('phone_new', sa.String(length=20), nullable=True))

    # Создаем уникальный индекс CONCURRENTLY
    # Alembic не поддерживает CONCURRENTLY напрямую для op.create_unique_constraint,
    # поэтому используем op.execute для прямого SQL.
    op.execute('CREATE UNIQUE INDEX CONCURRENTLY ix_users_phone_new ON users (phone_new);')

def downgrade():
    op.drop_index('ix_users_phone_new', table_name='users')
    op.drop_column('users', 'phone_new')
    op.drop_column('users', 'last_name')
    op.drop_column('users', 'first_name')

Шаг 1.4: Добавление NOT NULL ограничения на phone_new

# alembic/versions/yyyyyyyyyyyy_make_phone_new_not_null.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Убеждаемся, что все phone_new не NULL перед применением NOT NULL
    # Это должно быть сделано скриптом переноса данных.
    # Если скрипт не был запущен или пропустил что-то, эта операция может упасть.
    # В реальной жизни, здесь может быть проверка или явное обновление
    # для оставшихся NULL, если это приемлемо.
    # Например: op.execute("UPDATE users SET phone_new = 'temp_missing_' || id WHERE phone_new IS NULL;")
    op.alter_column('users', 'phone_new',
               existing_type=sa.String(length=20),
               nullable=False,
               existing_nullable=True)

    # Добавляем UNIQUE ограничение на phone_new.
    # Так как индекс уже создан CONCURRENTLY, эта операция должна быть быстрой.
    # Мы добавляем его как NOT VALID, а затем валидируем.
    op.create_unique_constraint(op.f('uq_users_phone_new'), 'users', ['phone_new'], postgresql_deferrable=False)
    # op.execute('ALTER TABLE users ADD CONSTRAINT uq_users_phone_new UNIQUE (phone_new) NOT VALID;')
    # op.execute('ALTER TABLE users VALIDATE CONSTRAINT uq_users_phone_new;')


def downgrade():
    op.drop_constraint(op.f('uq_users_phone_new'), 'users', type_='unique')
    op.alter_column('users', 'phone_new',
               existing_type=sa.String(length=20),
               nullable=True,
               existing_nullable=False)

Примечание: Alembic op.create_unique_constraint не поддерживает NOT VALID напрямую. Если нужен NOT VALID и VALIDATE, то придется использовать op.execute для прямого SQL. Я оставил op.create_unique_constraint как более "Alembic-way", но в случае очень строгих требований к блокировкам, op.execute с NOT VALID и VALIDATE может быть предпочтительнее. Однако, поскольку индекс уже создан CONCURRENTLY, добавление NOT NULL и UNIQUE (как ограничение) должно быть относительно быстрым.

Шаг 2.1: Переименование столбцов и удаление старых

# alembic/versions/zzzzzzzzzzzz_rename_and_drop_old_columns.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Переименовываем phone_new в phone
    op.alter_column('users', 'phone_new', new_column_name='phone',
               existing_type=sa.String(length=20),
               existing_nullable=False) # Теперь phone_new уже NOT NULL

    # Удаляем старый столбец phone
    op.drop_column('users', 'phone_old') # Предполагаем, что старый phone был переименован в phone_old на предыдущем шаге,
 # или просто удаляем 'phone' если он не использовался приложением.
 # В нашем плане, мы просто удаляем 'phone'.
    op.drop_column('users', 'phone') # Удаляем оригинальный столбец phone

    # Удаляем full_name
    op.drop_column('users', 'full_name')

    # Переименовываем индекс, если нужно
    op.execute('ALTER INDEX ix_users_phone_new RENAME TO ix_users_phone;')
    op.execute('ALTER CONSTRAINT uq_users_phone_new RENAME TO uq_users_phone;')


def downgrade():
    # Восстанавливаем full_name
    op.add_column('users', sa.Column('full_name', sa.String(length=255), nullable=True))
    # Восстанавливаем старый phone
    op.add_column('users', sa.Column('phone', sa.String(length=20), nullable=True))

    # Переименовываем phone обратно в phone_new
    op.alter_column('users', 'phone', new_column_name='phone_new',
               existing_type=sa.String(length=20),
               existing_nullable=False)

    # Восстанавливаем NOT NULL и UNIQUE на phone_new
    op.alter_column('users', 'phone_new',
               existing_type=sa.String(length=20),
               nullable=False,
               existing_nullable=True)
    op.create_unique_constraint(op.f('uq_users_phone_new'), 'users', ['phone_new'])

    # Переименовываем индекс обратно
    op.execute('ALTER INDEX ix_users_phone RENAME TO ix_users_phone_new;')
    op.execute('ALTER CONSTRAINT uq_users_phone RENAME TO uq_users_phone_new;')

    # Здесь нужно будет восстановить данные full_name из first_name/last_name и phone из phone_new
    # Это очень сложная операция при откате, поэтому важно хорошо протестировать upgrade.
    # Для простоты, в downgrade мы не восстанавливаем данные, только схему.

Коррекция для Шага 2.1 (Downgrade): Downgrade для этого шага очень сложен, так как предполагает восстановление данных из удаленных столбцов. В реальной жизни, если full_name и старый phone были удалены, их восстановление потребует либо резервной копии, либо сложной логики. Поэтому, критически важно, чтобы upgrade был протестирован и стабилен.

---

3. Скрипт переноса существующих данных батчами

Этот скрипт будет запускаться после Шага 1.1 и Шага 1.3.

import psycopg2
import os
import time
from typing import Optional

# Конфигурация базы данных
DB_HOST = os.getenv('DB_HOST', 'localhost')
DB_NAME = os.getenv('DB_NAME', 'your_db')
DB_USER = os.getenv('DB_USER', 'your_user')
DB_PASSWORD = os.getenv('DB_PASSWORD', 'your_password')

BATCH_SIZE = 25000  # Оптимальный размер батча, можно настроить
PAUSE_BETWEEN_BATCHES_MS = 200 # Пауза в миллисекундах

def get_db_connection():
    return psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)

def parse_full_name(full_name: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    if not full_name:
        return None, None
    parts = full_name.split(' ', 1)
    first_name = parts[0] if parts else None
    last_name = parts[1] if len(parts) > 1 else None
    return first_name, last_name

def migrate_data_batch(conn, cursor, start_id: int, end_id: int):
    # Фаза 1: Перенос full_name -> first_name, last_name и phone -> phone_new
    # Для phone_new, если phone IS NULL, оставляем phone_new IS NULL на этом шаге
    query = """
        UPDATE users
        SET
            first_name = COALESCE(split_part(full_name, ' ', 1), first_name),
            last_name = COALESCE(substring(full_name from position(' ' in full_name) + 1), last_name),
            phone_new = COALESCE(phone, phone_new)
        WHERE id >= %s AND id < %s AND (full_name IS NOT NULL OR phone IS NOT NULL);
    """
    cursor.execute(query, (start_id, end_id))
    return cursor.rowcount

def fill_missing_phone_new_batch(conn, cursor, start_id: int, end_id: int):
    # Фаза 2: Заполнение phone_new для тех, у кого он NULL
    # Используем временное уникальное значение
    query = """
        UPDATE users
        SET phone_new = 'temp_missing_' || id::text
        WHERE id >= %s AND id < %s AND phone_new IS NULL;
    """
    cursor.execute(query, (start_id, end_id))
    return cursor.rowcount

def run_migration():
    conn = None
    try:
        conn = get_db_connection()
        conn.autocommit = False # Управляем транзакциями вручную
        cursor = conn.cursor()

        # Определяем диапазон ID
        cursor.execute("SELECT MIN(id), MAX(id) FROM users;")
        min_id, max_id = cursor.fetchone()

        if min_id is None or max_id is None:
            print("Таблица users пуста.")
            return

        print(f"Начинаем миграцию данных для ID от {min_id} до {max_id}")

        current_id = min_id
        while current_id <= max_id:
            batch_end_id = current_id + BATCH_SIZE
            print(f"Обрабатываем батч: ID от {current_id} до {batch_end_id - 1}")

            # Шаг 1: Перенос full_name и phone
            rows_updated_phase1 = migrate_data_batch(conn, cursor, current_id, batch_end_id)
            print(f"  Фаза 1 (full_name/phone): Обновлено {rows_updated_phase1} строк.")

            # Шаг 2: Заполнение отсутствующих phone_new (если phone был NULL)
            rows_updated_phase2 = fill_missing_phone_new_batch(conn, cursor, current_id, batch_end_id)
            print(f"  Фаза 2 (missing phone_new): Обновлено {rows_updated_phase2} строк.")

            conn.commit()
            current_id = batch_end_id

            if current_id <= max_id:
                time.sleep(PAUSE_BETWEEN_BATCHES_MS / 1000.0) # Пауза в секундах

        print("Миграция данных завершена.")

    except Exception as e:
        print(f"Ошибка во время миграции: {e}")
        if conn:
            conn.rollback()
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    run_migration()

Размер батча и паузы:

• BATCH_SIZE = 25000: Для 18 млн строк, это около 720 батчей. Хорошая отправная точка. Если UPDATE занимает слишком много времени (более 100-200 мс), уменьшите размер батча. Если слишком быстро, можно увеличить.
• PAUSE_BETWEEN_BATCHES_MS = 200: Пауза в 200 мс (0.2 секунды) между батчами. Это помогает снизить пиковую нагрузку на базу данных и предотвратить перегрузку. Если база все еще испытывает стресс, увеличьте паузу. Если нагрузка низкая, можно уменьшить.
• Мониторинг: Во время выполнения скрипта обязательно мониторьте нагрузку на CPU, IOPS, задержки запросов и количество активных/ожидающих блокировок на PostgreSQL.

---

4. Какие операции в этой СУБД берут тяжёлые блокировки и как их избежать

| Операция | Тип блокировки (PostgreSQL) | Как избежать/минимизировать