Go 数据库迁移工具链：从手写 SQL 到自动化版本管理的演进实践

一、手写 SQL 迁移的熵增：当add_column_migration_v3_final_final2.sql成为常规命名

数据库 Schema 变更是应用演进中最危险的操作。手写 SQL 的方式来管理这些变更，经过一段时间后通常会变成：一堆数字编号的 SQL 文件散落在migrations/目录中、文件名不能反映内容、不知道哪些已执行哪些未执行、回滚需要手动写反向 SQL。

更危险的是多环境管理——开发环境的迁移状态和生产环境不一致，导致某些迁移在开发环境测试过但在生产环境执行失败。

数据库迁移工具链解决了这些混乱。它的核心思想和 Git 一样：每次变更都是一个有历史记录的版本，可以前进（Up），也可以回滚（Down）。

graph TB subgraph Migration[迁移文件] M1[001_create_users.sql] M2[002_add_email_column.sql] M3[003_create_orders.sql] end subgraph DB[数据库] T[migrations 表<br/>记录已执行的版本] end M1 --> A[Apply: 执行 M1] A --> T M2 --> B[Apply: 执行 M2] B --> T M3 --> C[Apply: 执行 M3] C --> T subgraph Rollback[回滚] R1[Reverse: 回滚 M3] R2[Reverse: 回滚 M2] T --> R1 R1 --> R2 end style T fill:#4dabf7,color:#fff style Rollback fill:#ffd43b,color:#000

二、迁移版本的追踪机制：版本表与幂等性保证

迁移工具的核心是一个schema_migrations表。它通常只有两个字段：version（版本号）和applied_at（执行时间）。每个迁移文件以版本号为前缀命名（如001_create_users.up.sql），工具读取文件列表并与表中已执行的版本对比，计算出待执行的迁移。

关键设计：幂等性。同一个迁移不应该被执行两次。工具通过版本号保证：如果版本号已在schema_migrations中存在，跳过该迁移。这要求迁移文件本身也应该是幂等的——使用IF NOT EXISTS或先检查再操作。

三、从零构建 Go 迁移工具

package migration import ( "database/sql" "fmt" "log" "os" "path/filepath" "sort" "strings" "time" ) // Migration 表示一个迁移 type Migration struct { Version string Description string UpSQL string DownSQL string } // Runner 迁移执行器 type Runner struct { db *sql.DB migrations []Migration tableName string } func NewRunner(db *sql.DB, tableName string) *Runner { return &Runner{ db: db, tableName: tableName, } } // Init 创建迁移记录表 func (r *Runner) Init() error { _, err := r.db.Exec(fmt.Sprintf(` CREATE TABLE IF NOT EXISTS %s ( version VARCHAR(14) PRIMARY KEY, description TEXT NOT NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) `, r.tableName)) return err } // LoadFromDir 从目录加载迁移文件 func (r *Runner) LoadFromDir(dir string) error { entries, err := os.ReadDir(dir) if err != nil { return fmt.Errorf("read migration dir: %w", err) } migrationMap := make(map[string]*Migration) for _, entry := range entries { name := entry.Name() // 解析文件名: {version}_{description}.{up|down}.sql parts := strings.SplitN(name, "_", 2) if len(parts) < 2 { continue } version := parts[0] rest := parts[1] if _, ok := migrationMap[version]; !ok { migrationMap[version] = &Migration{Version: version} } content, err := os.ReadFile(filepath.Join(dir, name)) if err != nil { return fmt.Errorf("read %s: %w", name, err) } if strings.HasSuffix(name, ".up.sql") { desc := strings.TrimSuffix(rest, ".up.sql") migrationMap[version].Description = desc migrationMap[version].UpSQL = string(content) } else if strings.HasSuffix(name, ".down.sql") { migrationMap[version].DownSQL = string(content) } } // 按版本号排序 versions := make([]string, 0, len(migrationMap)) for v := range migrationMap { versions = append(versions, v) } sort.Strings(versions) for _, v := range versions { r.migrations = append(r.migrations, *migrationMap[v]) } return nil } // Up 执行所有未应用的迁移 func (r *Runner) Up() error { applied, err := r.getAppliedVersions() if err != nil { return err } appliedSet := make(map[string]bool, len(applied)) for _, v := range applied { appliedSet[v] = true } for _, m := range r.migrations { if appliedSet[m.Version] { log.Printf("Skipping %s (already applied)", m.Version) continue } if m.UpSQL == "" { return fmt.Errorf("migration %s has no Up SQL", m.Version) } log.Printf("Applying %s: %s", m.Version, m.Description) tx, err := r.db.Begin() if err != nil { return fmt.Errorf("begin tx for %s: %w", m.Version, err) } if _, err := tx.Exec(m.UpSQL); err != nil { tx.Rollback() return fmt.Errorf("apply %s: %w", m.Version, err) } if _, err := tx.Exec( fmt.Sprintf("INSERT INTO %s (version, description) VALUES (?, ?)", r.tableName), m.Version, m.Description, ); err != nil { tx.Rollback() return fmt.Errorf("record %s: %w", m.Version, err) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit %s: %w", m.Version, err) } } return nil } // Down 回滚最近 N 个迁移 func (r *Runner) Down(steps int) error { applied, err := r.getAppliedVersions() if err != nil { return err } if steps > len(applied) { steps = len(applied) } toRollback := applied[len(applied)-steps:] // 建立版本号到迁移的映射 migrationMap := make(map[string]Migration) for _, m := range r.migrations { migrationMap[m.Version] = m } // 逆序回滚 for i := len(toRollback) - 1; i >= 0; i-- { version := toRollback[i] m, ok := migrationMap[version] if !ok { return fmt.Errorf("migration %s not found in files", version) } if m.DownSQL == "" { return fmt.Errorf("migration %s has no Down SQL", version) } log.Printf("Rolling back %s: %s", version, m.Description) tx, err := r.db.Begin() if err != nil { return fmt.Errorf("begin tx for %s: %w", version, err) } if _, err := tx.Exec(m.DownSQL); err != nil { tx.Rollback() return fmt.Errorf("rollback %s: %w", version, err) } if _, err := tx.Exec( fmt.Sprintf("DELETE FROM %s WHERE version = ?", r.tableName), version, ); err != nil { tx.Rollback() return fmt.Errorf("remove record %s: %w", version, err) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit rollback %s: %w", version, err) } } return nil } // Status 显示迁移状态 func (r *Runner) Status() error { applied, _ := r.getAppliedVersions() appliedSet := make(map[string]bool) for _, v := range applied { appliedSet[v] = true } for _, m := range r.migrations { status := "Pending" if appliedSet[m.Version] { status = "Applied" } fmt.Printf("[%s] %s - %s\n", status, m.Version, m.Description) } return nil } func (r *Runner) getAppliedVersions() ([]string, error) { rows, err := r.db.Query( fmt.Sprintf("SELECT version FROM %s ORDER BY version ASC", r.tableName), ) if err != nil { return nil, fmt.Errorf("query versions: %w", err) } defer rows.Close() var versions []string for rows.Next() { var v string if err := rows.Scan(&v); err != nil { return nil, err } versions = append(versions, v) } return versions, rows.Err() }

使用示例：

func main() { db, _ := sql.Open("postgres", "postgres://...") runner := migration.NewRunner(db, "schema_migrations") runner.Init() runner.LoadFromDir("./migrations") // 执行迁移 runner.Up() // 查看状态 runner.Status() // 回滚最近 1 个迁移 runner.Down(1) }

迁移文件示例：

-- migrations/001_create_users.up.sql CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- migrations/001_create_users.down.sql DROP TABLE IF EXISTS users;

四、迁移工具与 ORM 的选择

有些 ORM（如 GORM、Prisma）自带迁移功能。是否应该用它们？

使用 ORM 迁移：当项目已经使用该 ORM，迁移只是 ORM 的一个附带功能。优点是零额外配置。缺点是迁移能力受限于 ORM（如无法执行原生 SQL 的高级操作）。

使用独立迁移工具：当项目不使用 ORM、或者 ORM 的迁移能力不够时。优势是完全控制 SQL，可以进行复杂的变更（如数据迁移、索引重建）。缺点是需要手动维护。

推荐策略：无论是否使用 ORM，迁移文件都应该是纯 SQL。ORM 自动生成的迁移应该导出为 SQL 文件、审查后提交。迁移的决策权在于开发者，不在于工具。

五、总结

数据库迁移工具链的价值是用版本管理的方式管理 Schema 变更。Up/Down 模式让每次变更都可追溯、可回滚。事务保证让迁移在失败时自动恢复。

落地路径：先在项目中建立migrations/目录和版本号命名规范；然后引入或自建迁移工具，保证事务和幂等性；最后在 CI 中集成迁移检查（迁移状态、迁移数量）。

少即是多。迁移工具不需要复杂——版本号 + Up/Down SQL + 事务执行，这三个要素就是迁移管理的全部。