给 3000 万行的表加个字段：不停机数据库迁移的完整清单

# migration
migrations.AddField(
    model_name='downloadtask',
    name='priority',
    field=models.SmallIntegerField(default=0, verbose_name='优先级'),
)

ALTER TABLE "downloader_downloadtask" ADD COLUMN "priority" smallint DEFAULT 0 NOT NULL;
ALTER TABLE "downloader_downloadtask" ALTER COLUMN "priority" DROP DEFAULT;

# 默认值是可变函数 —— 必然重写全表
field=models.DateTimeField(default=timezone.now)        # Django 里这是 Python 默认值，不是 SQL DEFAULT
field=models.UUIDField(default=uuid.uuid4)

# 危险：可变默认值，重写全表
migrations.RunSQL('ALTER TABLE t ADD COLUMN created_at timestamptz DEFAULT now() NOT NULL')

# 第一步：加可空字段（毫秒级，安全）
migrations.AddField(
    model_name='downloadtask',
    name='created_at',
    field=models.DateTimeField(null=True),
)
# 第二步：数据迁移分批回填（见第六节）
# 第三步：加非空约束（见下一节）

1. 加可空字段              → 毫秒，安全
2. 分批回填历史数据        → 可控，随时停
3. 加非空约束              → 关键：怎么加决定了锁多久

-- 错误做法（扫全表且全程 ACCESS EXCLUSIVE）：
ALTER TABLE t ALTER COLUMN col SET NOT NULL;

-- 正确做法：分两步
-- 3a. 加一个「不立即验证」的检查约束（瞬时，只改元数据）
ALTER TABLE t ADD CONSTRAINT col_not_null CHECK (col IS NOT NULL) NOT VALID;

-- 3b. 验证已有数据（只拿 SHARE UPDATE EXCLUSIVE，不阻塞读写）
ALTER TABLE t VALIDATE CONSTRAINT col_not_null;

-- 1. 加可空字段
ALTER TABLE t ADD COLUMN priority smallint;

-- 2. 回填（分批，见第六节）

-- 3. 加 NOT VALID 约束（毫秒）
ALTER TABLE t ADD CONSTRAINT priority_not_null CHECK (priority IS NOT NULL) NOT VALID;

-- 4. 验证（扫全表但不阻塞读写，可以慢慢跑）
ALTER TABLE t VALIDATE CONSTRAINT priority_not_null;

-- 5. 设为 NOT NULL（PG 12+ 此时是瞬时的）
ALTER TABLE t ALTER COLUMN priority SET NOT NULL;

-- 6. 可选：删掉冗余的 CHECK（PG 12+ 里 SET NOT NULL 之后 CHECK 就多余了）
ALTER TABLE t DROP CONSTRAINT priority_not_null;

from django.db import migrations


class Migration(migrations.Migration):
    atomic = False          # ← 必须！CONCURRENTLY 不能在事务里跑

    dependencies = [('downloader', '0012_xxx')]

    operations = [
        migrations.RunSQL(
            sql='CREATE INDEX CONCURRENTLY idx_task_status_created '
                'ON downloader_downloadtask (status, created_at DESC)',
            reverse_sql='DROP INDEX CONCURRENTLY IF EXISTS idx_task_status_created',
        ),
    ]

ERROR: CREATE INDEX CONCURRENTLY cannot run inside a transaction block

-- 查无效索引
SELECT indexrelid::regclass AS idx, indisvalid
FROM pg_index
WHERE NOT indisvalid;

-- 清理（也要 CONCURRENTLY）
DROP INDEX CONCURRENTLY idx_xxx;

# settings.py（仅迁移时用，或者常开）
if env('ZERO_DOWNTIME_MIGRATIONS', default=False):
    DATABASES['default']['ENGINE'] = 'django_zero_downtime_migrations.backends.postgres'

阶段 1 Expand（扩展）：加新结构，新旧并存
        ↓ 部署代码：同时写新旧（双写），读旧的
阶段 2 Migrate（迁移）：回填历史数据，切读到新结构
        ↓ 部署代码：只读新的
阶段 3 Contract（收缩）：删掉旧结构

-- 危险：重写全表，全程 ACCESS EXCLUSIVE
ALTER TABLE t ALTER COLUMN size TYPE bigint;

-- 1. 加新列
ALTER TABLE t ADD COLUMN size_new bigint;

-- 2. 双写（代码里同时写 size 和 size_new）+ 回填历史数据（分批）

-- 3. 切换（这一步要短，但仍是 ACCESS EXCLUSIVE；可以在低峰做）
BEGIN;
ALTER TABLE t RENAME COLUMN size TO size_old;
ALTER TABLE t RENAME COLUMN size_new TO size;
COMMIT;

-- 4. 观察几天，确认没问题后
ALTER TABLE t DROP COLUMN size_old;

# 危险：一个巨大的事务，锁大量行，产生大量 WAL，可能撑爆磁盘/复制延迟
def fill_priority(apps, schema_editor):
    DownloadTask = apps.get_model('downloader', 'DownloadTask')
    DownloadTask.objects.filter(priority__isnull=True).update(priority=0)

import time
from django.db import connection


def batch_fill(model_name, field, value, batch_size=2000, sleep=0.05):
    """分批回填：每批一个小事务，可随时中断并从中断处继续。"""
    Model = apps.get_model('downloader', model_name)
    total = 0
    while True:
        # 取出一批 ID（用子查询限制，避免大 OFFSET）
        ids = list(
            Model.objects.filter(**{f'{field}__isnull': True})
            .values_list('id', flat=True)[:batch_size]
        )
        if not ids:
            break
        # 只更新这批（小事务）
        Model.objects.filter(id__in=ids).update(**{field: value})
        total += len(ids)
        print(f'filled {total}')
        time.sleep(sleep)      # 给主库和 vacuum 喘口气
    return total

30000000 / 2000 = 15000 批 × 80ms = 1200 秒 = 20 分钟

SET lock_timeout = '3s';
SET statement_timeout = '300s';

DDL 在等待 ACCESS EXCLUSIVE 锁
   ↓
后续所有对这张表的查询（包括 SELECT！）都要排在 DDL 后面
   ↓
整张表的访问全部阻塞

ERROR:  canceling statement due to lock timeout

from django.db import migrations

SET_LOCK_TIMEOUT = """
SET lock_timeout = '3s';
SET statement_timeout = '600s';
"""


class Migration(migrations.Migration):
    atomic = False
    operations = [
        migrations.RunSQL(SET_LOCK_TIMEOUT),      # 先设超时
        migrations.RunSQL('CREATE INDEX CONCURRENTLY ...'),
    ]

ALTER ROLE deploy_user SET lock_timeout = '3s';
ALTER ROLE deploy_user SET statement_timeout = '600s';

1. 备份（大改动前，尤其是有数据回填的）
      ↓
2. 跑「向前兼容」的迁移（加表、加可空字段、CONCURRENTLY 加索引、NOT VALID 约束）
      ↓
3. 灰度发布代码（滚动切换，新旧并存）
      ↓
4. 跑数据回填（异步任务，可中断）
      ↓
5. 跑「验证类」迁移（VALIDATE CONSTRAINT）—— 不阻塞，可以慢慢来
      ↓
6. 观察 1~7 天
      ↓
7. 跑「清理类」迁移（删字段、删旧表）

[ ] 1. 确认生产 PostgreSQL 版本（10 和 11+ 的行为差异很大）
[ ] 2. 这次迁移涉及的表有多少行？（SELECT count(*) 或者用 reltuples 估算）
[ ] 3. 每条 DDL 拿什么锁？对照第一节的表
[ ] 4. 加索引用了 CONCURRENTLY 吗？migration 设 atomic=False 了吗？
[ ] 5. 加非空约束用了 NOT VALID + VALIDATE 吗？
[ ] 6. 有数据回填吗？是分批的吗？估算过多久？
[ ] 7. 有没有设 lock_timeout？
[ ] 8. 有没有破坏性变更（删字段/改类型）？如果有，拆成独立发布了吗？
[ ] 9. 在「和生产同量级数据」的测试库上跑过一遍并计时了吗？
[ ] 10. 大改动前有备份吗？（如果是不可逆操作）
[ ] 11. 回滚方案是什么？（如果新代码有问题，数据库兼容吗？）
[ ] 12. 安排在业务低峰了吗？