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

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 事务执行这三个要素就是迁移管理的全部。