Migration 版本管理

Schema 变更不能靠手工执行 SQL——需要版本化、可重放、可回滚的 Migration 系统。本节讲如何在主流框架（Rails / Django / Golang）中正确管理 PostgreSQL 的 Schema 变更历史，以及如何处理 CONCURRENTLY 等特殊场景。

Migration 的核心原则

✅ 好的 Migration 实践：
1. 每次变更都有对应的 Migration 文件
2. Migration 文件一旦提交不修改（幂等性）
3. 版本号按时间戳排序（防止冲突）
4. 每个 Migration 有对应的回滚（down migration）
5. 生产环境 Migration 在维护窗口或自动部署流水线中执行
❌ 坏的实践：
- 直接 SSH 到生产数据库手工执行 SQL
- 修改已执行的 Migration 文件
- 把多个不相关的变更放在一个 Migration

Rails：Active Record Migrations

# 普通 Migration（会在事务中执行）
class AddPaymentMethodToOrders < ActiveRecord::Migration[7.1]
def change
add_column :orders, :payment_method, :string
end
end
# ❌ 普通方式建索引（会锁表）
class AddIndexToOrdersUserId < ActiveRecord::Migration[7.1]
def change
add_index :orders, :user_id  # 等同于 CREATE INDEX，会阻塞写入
end
end
# ✅ CONCURRENTLY 需要禁用事务（Rails 特有）
class AddIndexConcurrentlyToOrdersUserId < ActiveRecord::Migration[7.1]
disable_ddl_transaction!  # 关键：CREATE INDEX CONCURRENTLY 不能在事务中运行
def change
add_index :orders, :user_id, algorithm: :concurrently
end
end
# ✅ 添加 NOT NULL 约束（分步安全方式）
class AddNotNullToOrdersPaymentMethod < ActiveRecord::Migration[7.1]
disable_ddl_transaction!
def up
# 步骤1：添加列（允许 NULL）
add_column :orders, :payment_method, :string unless column_exists?(:orders, :payment_method)
# 步骤2：批量回填（在 Rails 中用 in_batches）
Order.in_batches(of: 5000) do |batch|
batch.update_all(payment_method: 'unknown')
end
# 步骤3：添加 CHECK 约束（NOT VALID）
execute <<-SQL
ALTER TABLE orders
ADD CONSTRAINT payment_method_not_null
CHECK (payment_method IS NOT NULL) NOT VALID;
SQL
# 步骤4：验证（后台，允许并发）
execute "ALTER TABLE orders VALIDATE CONSTRAINT payment_method_not_null;"
end
def down
execute "ALTER TABLE orders DROP CONSTRAINT IF EXISTS payment_method_not_null;"
remove_column :orders, :payment_method if column_exists?(:orders, :payment_method)
end
end

Django：Database Migrations

# migrations/0042_add_payment_method.py
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('orders', '0041_previous_migration'),
]
operations = [
# 添加允许 NULL 的列（安全）
migrations.AddField(
model_name='order',
name='payment_method',
field=models.CharField(max_length=50, null=True),
),
]

# migrations/0043_backfill_payment_method.py
from django.db import migrations
def backfill_payment_method(apps, schema_editor):
Order = apps.get_model('orders', 'Order')
# 分批回填
batch_size = 5000
while True:
ids = list(
Order.objects.filter(payment_method__isnull=True)
.values_list('id', flat=True)[:batch_size]
)
if not ids:
break
Order.objects.filter(id__in=ids).update(payment_method='unknown')
def reverse_backfill(apps, schema_editor):
pass  # 不回滚数据
class Migration(migrations.Migration):
dependencies = [('orders', '0042_add_payment_method')]
operations = [
migrations.RunPython(backfill_payment_method, reverse_backfill),
]

# migrations/0044_add_index_concurrently.py
from django.db import migrations
from django.contrib.postgres.operations import CreateExtension
class Migration(migrations.Migration):
# Django 4.1+ 支持 atomic = False 来禁用事务
atomic = False
dependencies = [('orders', '0043_backfill_payment_method')]
operations = [
# Django 4.1+ 的 AddIndexConcurrently
migrations.AddIndex(
model_name='order',
index=models.Index(fields=['user_id', 'status'], name='orders_user_status_idx'),
# Django 自动处理 CONCURRENTLY
),
]

golang-migrate / golang 项目

// migrations/000042_add_payment_method.up.sql
ALTER TABLE orders ADD COLUMN IF NOT EXISTS payment_method TEXT;

-- migrations/000042_add_payment_method.down.sql
ALTER TABLE orders DROP COLUMN IF EXISTS payment_method;

-- migrations/000043_backfill_payment_method.up.sql
-- 注意：大量数据时这个 Migration 可能运行较长时间
DO $$
DECLARE
batch_size INTEGER := 5000;
updated_count INTEGER;
BEGIN
LOOP
UPDATE orders
SET payment_method = 'unknown'
WHERE id IN (
SELECT id FROM orders WHERE payment_method IS NULL LIMIT batch_size
);
GET DIAGNOSTICS updated_count = ROW_COUNT;
EXIT WHEN updated_count = 0;
PERFORM pg_sleep(0.05);  -- 稍作缓冲
END LOOP;
END;
$$;

-- migrations/000044_add_index_concurrently.up.sql
-- golang-migrate 默认每个文件在事务中执行
-- 需要用特殊注释告诉 golang-migrate 禁用事务
-- golang-migrate 0.14+ 支持：
-- migrate:begin-no-transaction
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_user_status
ON orders (user_id, status);
-- migrate:end-no-transaction

squitch：SQL-first Migration 管理

# Sqitch 是语言无关的 SQL Migration 管理工具
# 适合多语言团队或需要精细控制的场景
sqitch init myapp --engine pg
# 添加新变更
sqitch add add_payment_method --note "Add payment_method column to orders"
# 编辑 deploy/add_payment_method.sql
cat >> deploy/add_payment_method.sql << 'EOF'
BEGIN;
ALTER TABLE orders ADD COLUMN IF NOT EXISTS payment_method TEXT;
COMMIT;
EOF
# 编辑 revert/add_payment_method.sql
cat >> revert/add_payment_method.sql << 'EOF'
BEGIN;
ALTER TABLE orders DROP COLUMN IF EXISTS payment_method;
COMMIT;
EOF
# 部署
sqitch deploy db:pg://user:pass@host/dbname
# 回滚一步
sqitch revert --to @HEAD^ db:pg://user:pass@host/dbname
# 查看状态
sqitch status db:pg://user:pass@host/dbname

生产 Migration 流程规范

Migration 审查清单

📋 提交 Migration 前必须确认：
Schema 变更安全性：
- [ ] 是否在大表上做 ALTER COLUMN TYPE？（需要展开-迁移-切换）
- [ ] 是否添加 NOT NULL 约束？（使用 NOT VALID + VALIDATE）
- [ ] 是否创建索引？（使用 CONCURRENTLY）
- [ ] 是否有 disable_ddl_transaction! / atomic=False？（CONCURRENTLY 必须）
数据迁移：
- [ ] 批量操作是否分批？（每批 < 1 万行）
- [ ] 是否有回滚 SQL？
- [ ] 是否在 Staging 测试过耗时？
运维准备：
- [ ] 是否通知了 DBA/运维？
- [ ] 是否有对应的监控告警？
- [ ] 是否准备好回滚方案？

本书总结：PostgreSQL 学习路线回顾

基础安全运行：
✅ 第1章：安装、psql、pg_hba 认证
✅ 第2章：类型选择、主键策略、约束设计
查询能力：
✅ 第3章：索引类型、EXPLAIN ANALYZE、慢查询
✅ 第4章：CTE、窗口函数、Pivot
✅ 第5章：JSONB、GIN 索引、修改操作
生产可靠性：
✅ 第6章：MVCC、事务隔离、锁与死锁、乐观锁
✅ 第7章：全文搜索、中文分词、pgvector
高可用与运维：
✅ 第8章：pg_dump、流复制、Patroni
✅ 第9章：pgBouncer、参数调优、pg_stat 监控
✅ 第10章：大表变更、CONCURRENTLY、Migration 管理

恭喜完成 PostgreSQL 实战指南！ 你现在掌握了从安装配置到生产高可用的完整 PostgreSQL 技能体系。下一步推荐：把第3章的 EXPLAIN ANALYZE 用到你当前项目的最慢查询上，用本书第5章的 JSONB 技巧重构一个过度使用 MySQL 的 EAV 表，以及按第10章的 Migration 规范重审你团队的 Schema 变更流程。