数据库迁移管理

概览

实施健壮的数据库迁移系统，包括版本控制、回滚能力以及数据转换策略。包括迁移框架和生产部署模式。

何时使用

架构版本控制和演化
数据转换和清理
添加/移除表和列
索引创建和优化
迁移测试和验证
回滚计划和执行
多环境部署

迁移框架设置

PostgreSQL - 架构版本控制

-- 创建迁移跟踪表
CREATE TABLE schema_migrations (
  version BIGINT PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  duration_ms INTEGER,
  checksum VARCHAR(64)
);

-- 创建迁移日志表
CREATE TABLE migration_logs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  version BIGINT NOT NULL,
  status VARCHAR(20) NOT NULL,
  error_message TEXT,
  rolled_back_at TIMESTAMP,
  executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 函数记录迁移
CREATE OR REPLACE FUNCTION record_migration(
  p_version BIGINT,
  p_name VARCHAR,
  p_duration_ms INTEGER
) RETURNS void AS $$
BEGIN
  INSERT INTO schema_migrations (version, name, duration_ms)
  VALUES (p_version, p_name, p_duration_ms)
  ON CONFLICT (version) DO UPDATE
  SET executed_at = CURRENT_TIMESTAMP;
END;
$$ LANGUAGE plpgsql;

MySQL - 迁移跟踪

-- 创建MySQL的迁移表
CREATE TABLE schema_migrations (
  version BIGINT PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  duration_ms INT,
  checksum VARCHAR(64)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 迁移状态表
CREATE TABLE migration_status (
  id INT AUTO_INCREMENT PRIMARY KEY,
  version BIGINT NOT NULL,
  status ENUM('pending', 'completed', 'failed', 'rolled_back'),
  error_message TEXT,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

常见迁移模式

添加列

PostgreSQL - 安全添加列：

-- 迁移：20240115_001_add_phone_to_users.sql

-- 添加列并设置默认值（非阻塞）
ALTER TABLE users
ADD COLUMN phone VARCHAR(20) DEFAULT '';

-- 添加约束后填充
ALTER TABLE users
ADD CONSTRAINT phone_format
CHECK (phone = '' OR phone ~ '^\+?[0-9\-\(\)]{10,}$');

-- 创建索引
CREATE INDEX CONCURRENTLY idx_users_phone ON users(phone);

-- 回滚：
-- DROP INDEX CONCURRENTLY idx_users_phone;
-- ALTER TABLE users DROP COLUMN phone;

MySQL - 添加列：

-- 迁移：20240115_001_add_phone_to_users.sql

-- 使用ALTER添加列
ALTER TABLE users
ADD COLUMN phone VARCHAR(20) DEFAULT '',
ADD INDEX idx_phone (phone);

-- 回滚：
-- ALTER TABLE users DROP COLUMN phone;

重命名列

PostgreSQL - 列重命名：

-- 迁移：20240115_002_rename_user_name_columns.sql

-- 重命名列
ALTER TABLE users RENAME COLUMN user_name TO full_name;
ALTER TABLE users RENAME COLUMN user_email TO email_address;

-- 更新索引
REINDEX TABLE users;

-- 回滚：
-- ALTER TABLE users RENAME COLUMN email_address TO user_email;
-- ALTER TABLE users RENAME COLUMN full_name TO user_name;

创建索引非阻塞

PostgreSQL - 并发索引创建：

-- 迁移：20240115_003_add_performance_indexes.sql

-- 创建索引不阻塞写入
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders(user_id, created_at DESC);

CREATE INDEX CONCURRENTLY idx_products_category_active
ON products(category_id)
WHERE active = true;

-- 验证索引创建
SELECT schemaname, tablename, indexname, idx_scan
FROM pg_stat_user_indexes
WHERE indexname LIKE 'idx_%';

-- 回滚：
-- DROP INDEX CONCURRENTLY idx_orders_user_created;
-- DROP INDEX CONCURRENTLY idx_products_category_active;

MySQL - 在线索引创建：

-- 迁移：20240115_003_add_performance_indexes.sql

-- 使用ALGORITHM=INPLACE和LOCK=NONE创建索引
ALTER TABLE orders
ADD INDEX idx_user_created (user_id, created_at),
ALGORITHM=INPLACE, LOCK=NONE;

-- 监控进度
SELECT * FROM INFORMATION_SCHEMA.PROCESSLIST
WHERE INFO LIKE 'ALTER TABLE%';

数据转换

PostgreSQL - 数据清理迁移：

-- 迁移：20240115_004_normalize_email_addresses.sql

-- 规范化现有电子邮件地址
UPDATE users
SET email = LOWER(TRIM(email))
WHERE email != LOWER(TRIM(email));

-- 通过保留最新的删除重复项
DELETE FROM users
WHERE id NOT IN (
  SELECT DISTINCT ON (LOWER(email)) id
  FROM users
  ORDER BY LOWER(email), created_at DESC
);

-- 回滚：从备份恢复（数据更改无安全回滚）

MySQL - 批量数据更新：

-- 迁移：20240115_004_update_product_categories.sql

-- 使用JOIN更新多行
UPDATE products p
JOIN category_mapping cm ON p.old_category = cm.old_name
SET p.category_id = cm.new_category_id
WHERE p.old_category IS NOT NULL;

-- 验证更新
SELECT COUNT(*) as updated_count
FROM products
WHERE category_id IS NOT NULL;

表结构变更

PostgreSQL - 修改表迁移：

-- 迁移：20240115_005_modify_order_columns.sql

-- 添加新列
ALTER TABLE orders
ADD COLUMN status_updated_at TIMESTAMP;

-- 添加约束
ALTER TABLE orders
ADD CONSTRAINT valid_status
CHECK (status IN ('pending', 'processing', 'completed', 'cancelled'));

-- 为现有记录设置默认值
UPDATE orders
SET status_updated_at = updated_at
WHERE status_updated_at IS NULL;

-- 将列设置为NOT NULL
ALTER TABLE orders
ALTER COLUMN status_updated_at SET NOT NULL;

-- 回滚：
-- ALTER TABLE orders DROP COLUMN status_updated_at;
-- ALTER TABLE orders DROP CONSTRAINT valid_status;

测试迁移

PostgreSQL - 事务中测试：

-- 在事务中测试迁移（将被回滚）
BEGIN;

-- 运行迁移语句
ALTER TABLE users ADD COLUMN test_column VARCHAR(255);

-- 验证数据
SELECT COUNT(*) FROM users;
SELECT COUNT(DISTINCT email) FROM users;

-- 如果发现问题则回滚
ROLLBACK;

-- 或者如果一切顺利则提交
COMMIT;

验证迁移：

-- 检查迁移是否已应用
SELECT version, name, executed_at FROM schema_migrations
WHERE version = 20240115005;

-- 验证表结构
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = 'users'
ORDER BY ordinal_position;

回滚策略

PostgreSQL - 双向迁移：

-- 迁移文件：20240115_006_add_user_status.sql

-- ===== UP =====
CREATE TYPE user_status AS ENUM ('active', 'suspended', 'deleted');
ALTER TABLE users ADD COLUMN status user_status DEFAULT 'active';

-- ===== DOWN =====
-- ALTER TABLE users DROP COLUMN status;
-- DROP TYPE user_status;

回滚执行：

-- 函数回滚到特定版本
CREATE OR REPLACE FUNCTION rollback_to_version(p_target_version BIGINT)
RETURNS TABLE (version BIGINT, name VARCHAR, status VARCHAR) AS $$
BEGIN
  -- 按逆序执行下迁移
  RETURN QUERY
  SELECT m.version, m.name, 'rolled_back'::VARCHAR
  FROM schema_migrations m
  WHERE m.version > p_target_version
  ORDER BY m.version DESC;
END;
$$ LANGUAGE plpgsql;

生产部署

安全迁移清单：

在类似生产环境的数据库上测试迁移
迁移前验证备份存在
在低流量窗口安排
监控表锁和长时间运行的查询
准备好回滚计划
测试回滚程序
记录所有更改
尽可能在事务中运行
迁移后验证数据完整性
与迁移协调更新应用程序代码

PostgreSQL - 长事务安全：

-- 使用语句超时防止挂起迁移
SET statement_timeout = '30min';

-- 使用锁超时防止死锁
SET lock_timeout = '5min';

-- 使用超时运行迁移
ALTER TABLE large_table
ADD COLUMN new_column VARCHAR(255),
ALGORITHM='INPLACE';

迁移示例

组合迁移 - 多项更改：

-- 迁移：20240115_007_refactor_user_tables.sql

BEGIN;

-- 1. 从旧列创建新列并填充数据
ALTER TABLE users ADD COLUMN full_name VARCHAR(255);
UPDATE users SET full_name = first_name || ' ' || last_name;

-- 2. 添加索引
CREATE INDEX idx_users_full_name ON users(full_name);

-- 3. 添加新约束
ALTER TABLE users
ADD CONSTRAINT email_unique UNIQUE(email);

-- 4. 验证后删除旧列
-- ALTER TABLE users DROP COLUMN first_name;
-- ALTER TABLE users DROP COLUMN last_name;

COMMIT;

数据库迁移管理 database-migration-management