Go 数据库迁移入门：SQL 文件、版本表和上线顺序

migrations/
  001_create_tasks.up.sql
  001_create_tasks.down.sql
  002_add_task_due_date.up.sql
  002_add_task_due_date.down.sql
  003_add_task_priority_index.up.sql
  003_add_task_priority_index.down.sql

-- 001_create_tasks.up.sql
CREATE TABLE tasks (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    title VARCHAR(255) NOT NULL,
    description TEXT,
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 001_create_tasks.down.sql
DROP TABLE IF EXISTS tasks;

CREATE TABLE schema_migrations (
    version BIGINT PRIMARY KEY,
    applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO schema_migrations (version) VALUES (1);

package main

import (
	"context"
	"database/sql"
)

func appliedVersions(ctx context.Context, db *sql.DB) (map[int]bool, error) {
	rows, err := db.QueryContext(ctx, `SELECT version FROM schema_migrations ORDER BY version`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	versions := map[int]bool{}
	for rows.Next() {
		var v int
		if err := rows.Scan(&v); err != nil {
			return nil, err
		}
		versions[v] = true
	}
	return versions, rows.Err()
}

-- 004_add_task_due_date.up.sql
ALTER TABLE tasks ADD COLUMN due_at TIMESTAMP NULL;

-- 005_make_due_date_not_null.up.sql
ALTER TABLE tasks MODIFY COLUMN due_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP;

-- 006_remove_old_deadline_field.up.sql
ALTER TABLE tasks DROP COLUMN old_deadline;

-- MySQL 8.0+ 支持在线添加索引
ALTER TABLE tasks ADD INDEX idx_status_created (status, created_at);

ALTER TABLE tasks ADD INDEX idx_status_created (status, created_at),
ALGORITHM=INPLACE, LOCK=NONE;

CREATE INDEX CONCURRENTLY idx_tasks_status_created ON tasks (status, created_at);

package main

import (
	"context"
	"database/sql"
	"fmt"
	"time"
)

type Migration struct {
	Version int
	Name    string
	Up      string
	Down    string
}

type Migrator struct {
	db *sql.DB
}

func (m *Migrator) RunMigration(ctx context.Context, migration Migration) error {
	// 检查是否已执行
	var exists bool
	err := m.db.QueryRowContext(ctx,
		"SELECT 1 FROM schema_migrations WHERE version = ?", migration.Version).Scan(&exists)
	if err == nil {
		// 已执行，跳过
		return nil
	}

	// 在事务中执行迁移
	tx, err := m.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	if _, err := tx.ExecContext(ctx, migration.Up); err != nil {
		return fmt.Errorf("exec up: %w", err)
	}

	if _, err := tx.ExecContext(ctx,
		"INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)",
		migration.Version, time.Now()); err != nil {
		return fmt.Errorf("record version: %w", err)
	}

	return tx.Commit()
}

-- 安全的向前兼容变更
ALTER TABLE tasks ADD COLUMN priority INT NULL DEFAULT 0;

-- 危险：可能锁住大量行
UPDATE tasks
SET due_at = DATE_ADD(created_at, INTERVAL 7 DAY)
WHERE due_at IS NULL;

package main

import (
	"context"
	"database/sql"
	"fmt"
	"time"
)

func BackfillDueAt(ctx context.Context, db *sql.DB, batchSize int) error {
	var lastID int64
	for {
		// 查询一批需要更新的记录
		rows, err := db.QueryContext(ctx, `
			SELECT id, created_at FROM tasks
			WHERE id > ? AND due_at IS NULL
			ORDER BY id
			LIMIT ?
		`, lastID, batchSize)
		if err != nil {
			return fmt.Errorf("query: %w", err)
		}

		type record struct {
			id        int64
			createdAt time.Time
		}
		var records []record
		for rows.Next() {
			var r record
			if err := rows.Scan(&r.id, &r.createdAt); err != nil {
				rows.Close()
				return fmt.Errorf("scan: %w", err)
			}
			records = append(records, r)
		}
		rows.Close()

		if len(records) == 0 {
			break // 全部处理完毕
		}

		// 逐条更新（可以优化为批量更新）
		for _, r := range records {
			dueAt := r.createdAt.Add(7 * 24 * time.Hour)
			if _, err := db.ExecContext(ctx,
				"UPDATE tasks SET due_at = ? WHERE id = ?", dueAt, r.id); err != nil {
				return fmt.Errorf("update id %d: %w", r.id, err)
			}
			lastID = r.id
		}

		// 中间状态可以随时中断，下次从 lastID 继续
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}
	}
	return nil
}

package main

import (
	"database/sql"
	"os"
	"testing"

	_ "github.com/go-sql-driver/mysql"
)

func TestMigrations(t *testing.T) {
	dsn := os.Getenv("TEST_DB_DSN")
	if dsn == "" {
		t.Skip("TEST_DB_DSN not set")
	}

	db, err := sql.Open("mysql", dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	// 1. 从空数据库开始，执行所有迁移
	migrator := &Migrator{db: db}
	migrations := loadMigrations() // 从目录加载所有迁移
	for _, m := range migrations {
		if err := migrator.RunMigration(m); err != nil {
			t.Fatalf("migration %d failed: %v", m.Version, err)
		}
	}

	// 2. 验证表结构是否符合预期
	var tableName string
	err = db.QueryRow("SHOW TABLES LIKE 'tasks'").Scan(&tableName)
	if err != nil {
		t.Fatal("tasks table not found after migrations")
	}

	// 3. 反向迁移（可选，测试回滚逻辑）
	for i := len(migrations) - 1; i >= 0; i-- {
		if err := migrator.RollbackMigration(migrations[i]); err != nil {
			t.Fatalf("rollback %d failed: %v", migrations[i].Version, err)
		}
	}
}

migrations/          # 结构迁移（所有环境共用）
  001_xxx.up.sql
  002_xxx.up.sql
seeds/               # 种子数据（仅本地/测试）
  users.sql
  tasks.sql

# 创建新迁移
migrate create -ext sql -dir migrations add_user_phone_column

# 执行所有待处理的迁移
migrate -path migrations -database "postgres://localhost/dbname?sslmode=disable" up

# 回滚最近 1 个迁移
migrate -path migrations -database "postgres://localhost/dbname?sslmode=disable" down 1

# 查看当前版本
migrate -path migrations -database "postgres://localhost/dbname?sslmode=disable" version