Function MigrateSchema

backend/migrator/migrator.go:44–127

MigrateSchema migrates the schema for metadata database.

(ctx context.Context, db *sql.DB)

// MigrateSchema migrates the schema for metadata database.
func MigrateSchema(ctx context.Context, db *sql.DB) error {
	// Acquire advisory lock to ensure only one replica runs migrations.
	// This blocks until the lock is available.
	lock, err := store.AcquireAdvisoryLock(ctx, db, store.AdvisoryLockKeyMigration)
	if err != nil {
		return errors.Wrap(err, "failed to acquire migration lock")
	}
	defer func() {
		if err := lock.Release(); err != nil {
			slog.Error("Failed to release migration advisory lock", log.BBError(err))
		}
	}()

	files, err := getSortedVersionedFiles()
	if err != nil {
		return err
	}

	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()

	var ok bool
	if err := conn.QueryRowContext(ctx,
		`SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'principal')`,
	).Scan(&ok); err != nil {
		return err
	}

	latestVersion := files[len(files)-1].version.String()
	// Initialize the latest schema.
	if !ok {
		buf, err := migrationFS.ReadFile(latestSchemaFileName)
		if err != nil {
			return errors.Wrapf(err, "failed to read latest schema %q", latestSchemaFileName)
		}
		if err := executeMigration(ctx, conn, string(buf), latestVersion); err != nil {
			return err
		}
		slog.Info(fmt.Sprintf("Initialized database schema with version %s.", latestVersion))
		return nil
	}

	latestDatabaseVersion, err := getLatestDatabaseVersion(ctx, conn)
	if err != nil {
		return err
	}
	if latestDatabaseVersion == nil {
		return errors.New("the latest database version is not found")
	}

	for _, f := range files {
		if f.version.LE(*latestDatabaseVersion) {
			continue
		}