SQLite 中的数据库版本控制：如何管理 Schema 变更？

引子：Schema变更管理的艺术与科学

作为最轻量级的关系型数据库，SQLite在嵌入式系统和移动应用中广受欢迎。但当数据库Schema需要变更时，如何实现平滑升级而不丢失数据？本文将用真实场景演示三种主流方案，并附赠最佳实践组合拳。

一、当Schema变更成为必修课

某社交APP用户量突破百万时，产品经理突发奇想："我们需要在用户表增加'个性签名'字段"。这简单的ALTER TABLE语句，在已发布的应用中却可能成为灾难：

• 旧版本APP遇到新版数据库会崩溃
• 用户数据可能因版本错乱而丢失
• 跨版本升级流程充满不确定性

（示例场景：从v1到v2的Schema变更）

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    username TEXT NOT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

# 需要升级到v2版本
ALTER TABLE users ADD COLUMN signature TEXT DEFAULT '';

二、版本控制三板斧

方案1：手工管理版本号（适合初创阶段）

import sqlite3

def get_db_version(conn):
    # 检查版本表是否存在
    conn.execute("CREATE TABLE IF NOT EXISTS db_version (version INTEGER)")
    return conn.execute("SELECT version FROM db_version").fetchone()[0] or 0

def migrate_v1_to_v2(conn):
    with conn:
        conn.execute("ALTER TABLE users ADD COLUMN signature TEXT DEFAULT ''")
        conn.execute("UPDATE db_version SET version = 2")

# 使用示例
conn = sqlite3.connect('app.db')
current_version = get_db_version(conn)
if current_version < 2:
    migrate_v1_to_v2(conn)

适用场景：个人项目/快速原型开发
暗礁警告：手工维护迁移顺序易出错，无回滚机制

方案2：迁移文件系统（中小项目推荐）

建立版本目录结构：

migrations/
├─ 001_initial.sql
├─ 002_add_signature.sql
└─ 003_add_indexes.sql

智能执行器实现：

import glob
import hashlib

class MigrationSystem:
    def __init__(self, conn):
        self.conn = conn
        self._ensure_version_table()
        
    def _ensure_version_table(self):
        self.conn.execute('''CREATE TABLE IF NOT EXISTS migrations (
            version INTEGER PRIMARY KEY,
            checksum TEXT NOT NULL
        )''')
    
    def run(self):
        executed = {row[0]: row[1] for row in 
                   self.conn.execute("SELECT version, checksum FROM migrations")}
        
        files = sorted(glob.glob('migrations/*.sql'), key=lambda x: int(x.split('_')[0]))
        for file in files:
            version = int(file.split('/')[-1].split('_')[0])
            with open(file) as f:
                sql = f.read()
                checksum = hashlib.md5(sql.encode()).hexdigest()
                
            if version not in executed:
                print(f'执行迁移: {file}')
                self.conn.executescript(sql)
                self.conn.execute("INSERT INTO migrations VALUES (?, ?)", 
                                (version, checksum))
            elif executed[version] != checksum:
                raise Exception(f"迁移文件{file}已被修改！")

核心优势：

1. 文件系统天然保证顺序
2. 校验码防止意外修改
3. 支持多人协作开发

方案3：ORM集成方案（大型项目优选）

使用Python SQLAlchemy示例：

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from alembic import op
import sqlalchemy as sa

Base = declarative_base()

# 声明数据模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False)
    created_at = Column(DateTime, server_default=sa.func.now())
    signature = Column(String(200))  # 新增字段

# 自动生成迁移脚本（alembic revision --autogenerate）
def upgrade():
    op.add_column('users', sa.Column('signature', sa.String(length=200)))

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

惊艳之处：

• 模型变更自动检测
• 生成升级/回滚双脚本
• 支持复杂数据类型

三、技术方案五维雷达图

从五个维度进行方案评测：

四、血泪换来的六条军规

1. 原子操作原则：每个迁移脚本必须是事务性的

with conn:  # 使用事务上下文
    conn.execute("ALTER TABLE...")
    conn.execute("UPDATE version...")

2. 版本跳跃陷阱：必须支持从任意旧版本逐级升级

-- 错误示例：仅支持v1→v3的升级
-- 正确做法：必须提供v1→v2和v2→v3的独立脚本

3. 数据迁移分离术：模式变更与数据迁移分开处理

/* 002_add_avatar.sql */
-- 模式变更
ALTER TABLE users ADD COLUMN avatar_url TEXT;

-- 单独创建数据迁移脚本
/* 002a_populate_avatars.sql */
UPDATE users SET avatar_url = 'default.png' WHERE avatar_url IS NULL;

4. 索引后置策略：先加字段再加索引

-- 错误顺序
CREATE INDEX idx_username ON users(username);  -- 已有数据量大时锁表

-- 正确流程
ALTER TABLE ADD COLUMN ...;
-- 数据填充完成后
CREATE INDEX CONCURRENTLY ...;  -- SQLite不支持但其他数据库需要注意

5. 多版本兼容魔法：通过视图实现新老版本兼容

-- 旧版APP看到的视图
CREATE VIEW legacy_users AS
SELECT id, username, created_at FROM users;

-- 新版APP使用真实表

6. 压测必杀技：变更前必须用真实数据量测试

# 生成百万测试数据
sqlite3 test.db < generate_test_data.sql
# 执行迁移并记录时间
time sqlite3 test.db < migrations/003_add_index.sql

五、未来战争：分布式场景的挑战

当应用发展到多节点架构时，经典方案开始捉襟见肘：

冲突场景：
节点A在v3运行时，节点B试图降级到v2

解药配方：
采用分布式锁方案

import redis
from contextlib import contextmanager

redis_conn = redis.Redis()

@contextmanager
def db_migration_lock():
    lock = redis_conn.lock('database_migration', timeout=300)
    if lock.acquire(blocking=True):
        try:
            yield
        finally:
            lock.release()
    else:
        raise Exception("无法获取迁移锁")

# 使用方式
with db_migration_lock():
    execute_migrations()