fix(restore): restore foreign dumps into a database of another name (#221)

README.md

@@ -193,6 +193,8 @@ databases:
 
 Dumps use `--single-transaction --routines --triggers --events --hex-blob --quick`, plus `--no-tablespaces --set-gtid-purged=OFF` on MySQL. A single-database dump has no `CREATE DATABASE`/`USE`, so it can be restored under another name.
 
+Restores into a named database (`-d`) also accept dumps written elsewhere: the `CREATE DATABASE`, `DROP DATABASE` and `USE` statements of a `mysqldump --databases` dump are skipped, so its objects land in the target database whatever the source was called (a dump of several databases is refused). `DEFINER` clauses are removed when the restoring user may not name another definer (`RESTORE_NO_DEFINER=auto`): views, triggers, routines and events then belong to that user. Dumps from MariaDB and from older MySQL versions restore into MySQL 8: the MariaDB sandbox line is dropped for the Oracle client, MariaDB's `utf8mb4_uca1400_ai_ci` (`_as_ci`, `_as_cs`) collations become `utf8mb4_0900_*`, and `sql_mode` values keep only the modes MySQL 8 knows (`NO_AUTO_CREATE_USER` goes). Data is never rewritten. On MySQL with binary logging on (the default), creating functions and triggers without `SUPER` also needs the server setting `log_bin_trust_function_creators=1`.
+
 The image is based on Debian 13 (trixie) slim. Oracle publishes its client for `amd64` and `arm64` only: on `arm/v7` the image has the MariaDB tools only, and MySQL servers are dumped with `mariadb-dump` (a warning is logged). `MYSQL_CLIENT=mysql|mariadb` forces one family.
 
 TLS: `DB_SSL_MODE=disabled` (default, as in 1.x) `| preferred | required | verify_ca | verify_identity`, with `DB_SSL_CA` (used for the verify modes), `DB_SSL_CERT`, `DB_SSL_KEY`. 1.x values `0` and `1` still mean disabled and required. With the MariaDB client, `verify_ca` also checks the host name.
@@ -423,6 +425,7 @@ Generated with `mysql-bkup config --env-docs`. Every secret also accepts `<NAME>
 | `TARGET_DB_SSL_MODE` |  | `disabled` |  | migrate: target TLS mode (same values as DB_SSL_MODE) |
 | `TARGET_DB_SSL_CA` |  | `/etc/ssl/certs/ca-certificates.crt` |  | migrate: target CA certificate file |
 | `MYSQL_CLIENT` |  | `auto` |  | Client tools: auto (mariadb-dump for MariaDB, Oracle mysqldump for MySQL), mysql or mariadb (auto, mysql, mariadb) |
+| `RESTORE_NO_DEFINER` |  | `auto` |  | Remove DEFINER clauses on restore (views, triggers, routines and events then belong to the restoring user); auto = when that user lacks SET_USER_ID/SUPER (auto, true, false) |
 
 
 ## Available image registries

docs/reference/index.md

@@ -249,6 +249,7 @@ Run `mysql-bkup config` to see the resolved configuration and where each value c
 | `TARGET_DB_SSL_MODE` |  | `disabled` |  | migrate: target TLS mode (same values as DB_SSL_MODE) |
 | `TARGET_DB_SSL_CA` |  | `/etc/ssl/certs/ca-certificates.crt` |  | migrate: target CA certificate file |
 | `MYSQL_CLIENT` |  | `auto` |  | Client tools: auto (mariadb-dump for MariaDB, Oracle mysqldump for MySQL), mysql or mariadb (auto, mysql, mariadb) |
+| `RESTORE_NO_DEFINER` |  | `auto` |  | Remove DEFINER clauses on restore (views, triggers, routines and events then belong to the restoring user); auto = when that user lacks SET_USER_ID/SUPER (auto, true, false) |
 
 
 ## Scheduled Backups

integration_test.go

@@ -540,6 +540,218 @@ func TestUploadsAndCorruption(t *testing.T) {
 	})
 }
 
+// dumpDatabases runs the server's own dump client (not mysql-bkup) with
+// --databases, as a dump made elsewhere and uploaded would be.
+func (s *dbServer) dumpDatabases(t *testing.T, dbs ...string) []byte {
+	t.Helper()
+	bin := "mysqldump"
+	if s.mariadb {
+		bin = "mariadb-dump"
+	}
+	args := append([]string{"exec", "-e", "MYSQL_PWD=" + rootPW, s.name, bin, "-uroot", "-h127.0.0.1",
+		"--routines", "--triggers", "--events", "--add-drop-database", "--databases"}, dbs...)
+	out, err := exec.Command("docker", args...).Output()
+	if err != nil {
+		t.Fatalf("%s: %v", bin, err)
+	}
+	return out
+}
+
+// mailer (re)creates the user mailer with privileges on dbs only (Miabi's
+// per-database users), recreating each database empty. Function and trigger
+// creation by non-SUPER users is allowed (see TestForeignDump).
+func (s *dbServer) mailer(t *testing.T, dbs ...string) {
+	t.Helper()
+	const user = "mailer"
+	pw := strings.NewReplacer(`\`, `\\`, `'`, `\'`).Replace(appPW)
+	q := fmt.Sprintf("DROP USER IF EXISTS '%[1]s'@'%%'; CREATE USER '%[1]s'@'%%' IDENTIFIED BY '%[2]s'; SET GLOBAL log_bin_trust_function_creators = 1;", user, pw)
+	for _, db := range dbs {
+		q += fmt.Sprintf(" DROP DATABASE IF EXISTS %[1]s; CREATE DATABASE %[1]s; GRANT ALL ON %[1]s.* TO '%[2]s'@'%%';", db, user)
+	}
+	s.mustSQL(t, q)
+}
+
+func serverOf(t *testing.T, mariadb bool) *dbServer {
+	t.Helper()
+	needDocker(t)
+	for _, s := range servers {
+		if s.mariadb == mariadb {
+			return s
+		}
+	}
+	t.Skip("no such server")
+	return nil
+}
+
+func writeFile(t *testing.T, dir, name string, b []byte) {
+	t.Helper()
+	if err := os.WriteFile(filepath.Join(dir, name), b, 0o644); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// TestCrossFlavourUploads: a mariadb-dump uploaded into MySQL (its sandbox
+// line, MariaDB definers) and a mysqldump uploaded into MariaDB, each into a
+// database of another name by a user with privileges on it only.
+func TestCrossFlavourUploads(t *testing.T) {
+	maria, mysql := serverOf(t, true), serverOf(t, false)
+	for _, c := range []struct {
+		name     string
+		src, dst *dbServer
+	}{
+		{"mariadb to mysql", maria, mysql},
+		{"mysql to mariadb", mysql, maria},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			c.src.loadFixture(t, "posta")
+			want := c.src.fingerprint(t, "posta")
+			c.dst.mailer(t, "mail")
+			store := setEnv(t, c.dst, "mailer", appPW)
+			writeFile(t, store, "posta.sql", c.src.dumpDatabases(t, "posta"))
+			out := mustRun(t, "restore", "-d", "mail", "-f", "posta.sql")
+			if got := c.dst.fingerprint(t, "mail"); got != want {
+				t.Fatalf("restored differs:\n got %s\nwant %s\n%s", got, want, out)
+			}
+		})
+	}
+}
+
+// classicDump is a hand-written dump in the style of the classicmodels sample
+// database: lowercase statements, USE without a semicolon, latin1.
+const classicDump = `/*!40101 SET NAMES utf8 */;
+/*!40101 SET SQL_MODE=''*/;
+create database if not exists ` + "`classicmodels`" + ` /*!40100 DEFAULT CHARACTER SET latin1 */;
+use classicmodels
+DROP TABLE IF EXISTS ` + "`customers`" + `;
+CREATE TABLE ` + "`customers`" + ` (
+  ` + "`customerNumber`" + ` int(11) NOT NULL,
+  ` + "`customerName`" + ` varchar(50) NOT NULL,
+  PRIMARY KEY (` + "`customerNumber`" + `)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1;
+insert  into ` + "`customers`" + `(` + "`customerNumber`,`customerName`" + `) values (103,'Atelier graphique'),(112,'Signal Gift Stores'),(114,'Australian Collectors, Co.');
+`
+
+// TestHandWrittenUpload: the classicmodels-style file, gzipped, uploaded into
+// a database of another name.
+func TestHandWrittenUpload(t *testing.T) {
+	forEachServer(t, func(t *testing.T, s *dbServer) {
+		s.mailer(t, "blog")
+		store := setEnv(t, s, "mailer", appPW)
+		var gz bytes.Buffer
+		zw := gzip.NewWriter(&gz)
+		_, _ = zw.Write([]byte(classicDump))
+		_ = zw.Close()
+		writeFile(t, store, "classicmodels.sql.gz", gz.Bytes())
+		mustRun(t, "restore", "-d", "blog", "-f", "classicmodels.sql.gz")
+		if got := s.mustSQL(t, "SELECT COUNT(*) FROM blog.customers"); got != "3" {
+			t.Fatalf("rows: %s", got)
+		}
+		if got := s.mustSQL(t, "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = 'classicmodels'"); got != "0" {
+			t.Fatal("the source database was created")
+		}
+	})
+}
+
+// TestDefinerModes: RESTORE_NO_DEFINER auto, true and false.
+func TestDefinerModes(t *testing.T) {
+	s := serverOf(t, false)
+	s.loadFixture(t, "posta")
+	// mysqldump cannot dump a view whose definer does not exist.
+	s.mustSQL(t, "DROP USER IF EXISTS 'ghost'@'%'; CREATE USER 'ghost'@'%'; CREATE DEFINER=`ghost`@`%` VIEW posta.v_ghost AS SELECT 1 AS one")
+	s.mailer(t, "mail_strict")
+	store := setEnv(t, s, "mailer", appPW)
+	writeFile(t, store, "posta.sql", s.dumpDatabases(t, "posta"))
+	viewDefiner := func(db string) string {
+		return s.mustSQL(t, "SELECT DEFINER FROM information_schema.VIEWS WHERE TABLE_SCHEMA = '"+db+"' AND TABLE_NAME = 'v_ghost'")
+	}
+
+	t.Setenv("RESTORE_NO_DEFINER", "false")
+	if out, code := run(t, "restore", "-d", "mail_strict", "-f", "posta.sql"); code == 0 {
+		t.Fatalf("restore with foreign definers as a database user succeeded:\n%s", out)
+	}
+
+	setEnv(t, s, "root", rootPW)
+	t.Setenv("STORAGE_PATH", store)
+	s.mustSQL(t, "DROP DATABASE IF EXISTS mail_auto; DROP DATABASE IF EXISTS mail_strip")
+	t.Setenv("RESTORE_NO_DEFINER", "auto")
+	mustRun(t, "restore", "-d", "mail_auto", "-f", "posta.sql")
+	if got := viewDefiner("mail_auto"); got != "ghost@%" {
+		t.Fatalf("auto as root: definer %q", got)
+	}
+	t.Setenv("RESTORE_NO_DEFINER", "true")
+	mustRun(t, "restore", "-d", "mail_strip", "-f", "posta.sql")
+	if got := viewDefiner("mail_strip"); !strings.HasPrefix(got, "root@") {
+		t.Fatalf("true as root: definer %q", got)
+	}
+}
+
+// TestMigrateToDatabaseUser: MySQL to MariaDB into a database of another name,
+// as a target user with privileges on it only (definers of the source).
+func TestMigrateToDatabaseUser(t *testing.T) {
+	src, dst := serverOf(t, false), serverOf(t, true)
+	src.loadFixture(t, "posta")
+	want := src.fingerprint(t, "posta")
+	dst.mailer(t, "mail")
+	setEnv(t, src, "root", rootPW)
+	t.Setenv("TARGET_DB_HOST", dst.name)
+	t.Setenv("TARGET_DB_USERNAME", "mailer")
+	t.Setenv("TARGET_DB_PASSWORD", appPW)
+	t.Setenv("TARGET_DB_NAME", "mail")
+	mustRun(t, "migrate", "-d", "posta")
+	if got := dst.fingerprint(t, "mail"); got != want {
+		t.Fatalf("migrated differs:\n got %s\nwant %s", got, want)
+	}
+}
+
+// TestForeignDump: a dump written elsewhere (mysqldump --databases: CREATE
+// DATABASE, USE, definers of the source server) uploaded into a database of
+// another name, by a user with privileges on that database only (Miabi).
+func TestForeignDump(t *testing.T) {
+	forEachServer(t, func(t *testing.T, s *dbServer) {
+		s.loadFixture(t, "posta")
+		want := s.fingerprint(t, "posta")
+		dump := func(dbs ...string) []byte { return s.dumpDatabases(t, dbs...) }
+		// With binary logging on (MySQL's default), only SUPER may create
+		// functions and triggers unless log_bin_trust_function_creators is set:
+		// a server setting, not something a restore can change.
+		s.mustSQL(t, fmt.Sprintf(`DROP USER IF EXISTS 'mailer'@'%%'; CREATE USER 'mailer'@'%%' IDENTIFIED BY '%s';
+DROP DATABASE IF EXISTS mail; CREATE DATABASE mail; GRANT ALL ON mail.* TO 'mailer'@'%%';
+DROP DATABASE IF EXISTS mail_root; SET GLOBAL log_bin_trust_function_creators = 1;`,
+			strings.NewReplacer(`\`, `\\`, `'`, `\'`).Replace(appPW)))
+		store := setEnv(t, s, "mailer", appPW)
+		if err := os.WriteFile(filepath.Join(store, "posta.sql"), dump("posta"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		out := mustRun(t, "restore", "-d", "mail", "-f", "posta.sql")
+		if got := s.fingerprint(t, "mail"); got != want {
+			t.Fatalf("restored differs:\n got %s\nwant %s\n%s", got, want, out)
+		}
+		if got := s.mustSQL(t, "SELECT DISTINCT DEFINER FROM information_schema.VIEWS WHERE TABLE_SCHEMA = 'mail'"); got != "mailer@%" {
+			t.Fatalf("view definer %q", got)
+		}
+		if got := s.fingerprint(t, "posta"); got != want {
+			t.Fatal("the source database was touched")
+		}
+
+		// root may keep the source definers (auto).
+		setEnv(t, s, "root", rootPW)
+		t.Setenv("STORAGE_PATH", store)
+		mustRun(t, "restore", "-d", "mail_root", "-f", "posta.sql")
+		if got := s.mustSQL(t, "SELECT DISTINCT DEFINER FROM information_schema.VIEWS WHERE TABLE_SCHEMA = 'mail_root'"); !strings.HasPrefix(got, "root@") {
+			t.Fatalf("definer not kept: %q", got)
+		}
+
+		// Two databases cannot go into one.
+		s.loadFixture(t, "other")
+		if err := os.WriteFile(filepath.Join(store, "two.sql"), dump("posta", "other"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		if out, code := run(t, "restore", "-d", "mail_two", "-f", "two.sql"); code == 0 || !strings.Contains(out, "more than one database") {
+			t.Fatalf("restore of two databases: exit %d\n%s", code, out)
+		}
+	})
+}
+
 // TestMiabiS3: Miabi's exact S3 argv and env against MinIO.
 func TestMiabiS3(t *testing.T) {
 	forEachServer(t, func(t *testing.T, s *dbServer) {

internal/mysql/engine.go

@@ -21,6 +21,9 @@ import (
 // Engine-specific settings.
 const (
 	SettingClient = "MYSQL_CLIENT"
+	// RestoreNoDefiner removes DEFINER clauses on restore: auto (when the
+	// restoring user may not name another definer), true or false.
+	RestoreNoDefiner = "RESTORE_NO_DEFINER"
 )
 
 // AllInOneName is the target (and default artifact) name of --all-in-one backups.
@@ -94,6 +97,12 @@ func (e *Engine) Settings(r *config.Registry) {
 		Group: config.GroupDatabase,
 		Usage: "Client tools: auto (mariadb-dump for MariaDB, Oracle mysqldump for MySQL), mysql or mariadb",
 	})
+	r.Add(config.Setting{
+		Key: RestoreNoDefiner, Default: "auto", Enum: []string{"auto", "true", "false"},
+		Group: config.GroupDatabase,
+		Usage: "Remove DEFINER clauses on restore (views, triggers, routines and events then belong " +
+			"to the restoring user); auto = when that user lacks SET_USER_ID/SUPER",
+	})
 }
 
 // Configure implements engine.Engine. It only validates values: commands that
@@ -373,14 +382,58 @@ func (e *Engine) Restore(ctx context.Context, t engine.Target, r io.Reader, info
 		return err
 	}
 	slog.Default().Info("Restoring database", "database", db, "file", info.File, "client", tl.name)
+	in := RestoreFilter(r, e.filterOptions(ctx, t.Values, c, tl, db))
+	defer func() { _ = in.Close() }()
 	return withOptionFile(t.Values, c, tl, func(cnf string) error {
 		return execx.Run(ctx, execx.Cmd{
 			Name: tl.path, Args: append([]string{"--defaults-extra-file=" + cnf}, args...),
-			Stdin: r, Log: slog.Default(), StderrLevel: slog.LevelWarn,
+			Stdin: in, Log: slog.Default(), StderrLevel: slog.LevelWarn,
 		})
 	})
 }
 
+// filterOptions describes how a dump restored into db through tl is adapted.
+func (e *Engine) filterOptions(ctx context.Context, v *config.Values, c Conn, tl tool, db string) FilterOptions {
+	o := FilterOptions{Target: db, NoDefiner: e.noDefiner(ctx, v, c, tl), OracleClient: !tl.mariadb}
+	if srv, err := e.server(ctx, v, c); err == nil {
+		o.MySQLServer = !srv.mariadb
+	}
+	return o
+}
+
+// noDefiner resolves RESTORE_NO_DEFINER for the restoring user.
+func (e *Engine) noDefiner(ctx context.Context, v *config.Values, c Conn, tl tool) bool {
+	switch v.String(RestoreNoDefiner) {
+	case "true":
+		return true
+	case "false":
+		return false
+	}
+	grants, err := query(ctx, v, c, tl, "SHOW GRANTS")
+	if err != nil {
+		slog.Default().Warn("Could not read the restoring user's grants; removing DEFINER clauses", "error", err)
+		return true
+	}
+	return !canSetDefiner(grants)
+}
+
+// canSetDefiner reports whether SHOW GRANTS output allows naming another
+// definer: SUPER, SET_USER_ID / SET_ANY_DEFINER (MySQL 8) or SET USER (MariaDB).
+func canSetDefiner(grants string) bool {
+	for _, line := range strings.Split(strings.ToUpper(grants), "\n") {
+		privs, _, global := strings.Cut(line, " ON *.* ")
+		if !global {
+			continue
+		}
+		for _, p := range []string{"ALL PRIVILEGES", "SUPER", "SET_USER_ID", "SET_ANY_DEFINER", "SET USER"} {
+			if strings.Contains(privs, p) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // MigrateTargets implements engine.Migrator: DB_NAME → TARGET_DB_NAME (default
 // the same name), or every database with --all-databases.
 func (e *Engine) MigrateTargets(ctx context.Context, v *config.Values) ([][2]engine.Target, error) {
@@ -454,6 +507,8 @@ func (e *Engine) Migrate(ctx context.Context, source, target engine.Target) erro
 		cancel()
 	}
 	pr, pw := io.Pipe()
+	in := RestoreFilter(pr, e.filterOptions(ctx, target.Values, tc, restTool, target.Name))
+	defer func() { _ = in.Close() }()
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -475,7 +530,7 @@ func (e *Engine) Migrate(ctx context.Context, source, target engine.Target) erro
 	err = withOptionFile(target.Values, tc, restTool, func(cnf string) error {
 		return execx.Run(ctx, execx.Cmd{
 			Name: restTool.path, Args: append([]string{"--defaults-extra-file=" + cnf}, restArgs...),
-			Stdin: pr, Log: slog.Default(), StderrLevel: slog.LevelWarn,
+			Stdin: in, Log: slog.Default(), StderrLevel: slog.LevelWarn,
 		})
 	})
 	if err != nil {

internal/mysql/restorefilter.go

@@ -0,0 +1,222 @@
+package mysql
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"regexp"
+	"strings"
+)
+
+// FilterOptions say how RestoreFilter adapts a dump.
+type FilterOptions struct {
+	// Target is the database the dump is restored into ("" for an
+	// all-in-one restore, which is left alone).
+	Target string
+	// NoDefiner removes DEFINER clauses.
+	NoDefiner bool
+	// OracleClient: Oracle's mysql client replays the dump.
+	OracleClient bool
+	// MySQLServer: the dump is restored into MySQL (not MariaDB).
+	MySQLServer bool
+}
+
+// RestoreFilter adapts a dump on its way to the client.
+//
+// With a Target (a restore into a named database), the dump's CREATE
+// DATABASE, DROP DATABASE and USE statements are dropped, so a dump taken with
+// --databases (or written by another tool) lands in target whatever its source
+// database was called. A dump naming two databases cannot go into one and
+// fails the restore.
+//
+// With NoDefiner, DEFINER clauses are removed from CREATE statements, so
+// views, triggers, routines and events are defined by the restoring user: only
+// a user with SET_USER_ID (SUPER) may name another one, and the source
+// accounts usually do not exist on the target anyway.
+//
+// With OracleClient, the MariaDB sandbox line that mariadb-dump writes
+// first ("/*M!999999\- enable the sandbox mode */") is dropped: Oracle's mysql
+// client rejects it, so a MariaDB dump could not be restored into MySQL. The
+// MariaDB client keeps it (it disables client shell commands for the dump).
+//
+// With MySQLServer, the utf8mb4_uca1400 collations MariaDB 11 uses by default
+// become their MySQL equivalents (utf8mb4_uca1400_ai_ci -> utf8mb4_0900_ai_ci),
+// outside data lines, and sql_mode values keep only the modes MySQL 8 knows
+// (NO_AUTO_CREATE_USER from MySQL 5.7 and MariaDB, MariaDB-only modes go), so
+// MariaDB and older MySQL dumps restore into MySQL 8.
+//
+// Lines are kept (an empty line replaces a dropped statement), so the client's
+// error line numbers still match the dump.
+func RestoreFilter(r io.Reader, o FilterOptions) io.ReadCloser {
+	pr, pw := io.Pipe()
+	go func() {
+		f := &restoreFilter{FilterOptions: o, delimiter: ";"}
+		err := f.run(r, pw)
+		if f.retargeted > 0 || f.definers > 0 || f.collations > 0 || f.sqlModes > 0 {
+			slog.Default().Info("Restore adapted to the target", "source_database", f.source, "database", o.Target,
+				"database_statements_skipped", f.retargeted, "definers_removed", f.definers,
+				"collations_mapped", f.collations, "sql_modes_adjusted", f.sqlModes)
+		}
+		_ = pw.CloseWithError(err)
+	}()
+	return pr
+}
+
+const ident = "(`(?:[^`]|``)+`|'(?:[^']|'')+'|\"(?:[^\"]|\"\")+\"|[\\w$]+)"
+
+var (
+	useRe       = regexp.MustCompile(`(?i)^USE\s+` + ident + `\s*;?$`)
+	createDBRe  = regexp.MustCompile(`(?i)^CREATE\s+(?:DATABASE|SCHEMA)\s*(?:/\*!\d*\s*IF\s+NOT\s+EXISTS\s*\*/|IF\s+NOT\s+EXISTS)?\s*` + ident + `[^;]*;$`)
+	dropDBRe    = regexp.MustCompile(`(?i)^(?:/\*!\d*\s*)?DROP\s+(?:DATABASE|SCHEMA)\s+(?:IF\s+EXISTS\s+)?` + ident + `\s*(?:\*/)?\s*;$`)
+	delimiterRe = regexp.MustCompile(`(?i)^DELIMITER\s+(\S+)`)
+	sandboxRe   = regexp.MustCompile(`^/\*M!999999\\- enable the sandbox mode \*/\s*$`)
+	definerPart = "(?:`(?:[^`]|``)*`|'(?:[^']|'')*'|\"(?:[^\"]|\"\")*\"|[\\w.$%-]+)"
+	// MariaDB collations with an exact MySQL 8 equivalent.
+	uca1400Re = regexp.MustCompile(`\butf8mb4_uca1400_(ai_ci|as_ci|as_cs)\b`)
+	dataRe    = regexp.MustCompile(`(?i)^(?:INSERT|REPLACE)\s`)
+	sqlModeRe = regexp.MustCompile(`(?i)(\bsql_mode\s*=\s*')([^']*)(')`)
+	definerRe = regexp.MustCompile(`(?i)\s*DEFINER\s*=\s*(?:CURRENT_USER(?:\s*\(\s*\))?|` + definerPart + `\s*@\s*` + definerPart + `|` + definerPart + `)`)
+)
+
+const lineBuf = 64 * 1024
+
+type restoreFilter struct {
+	FilterOptions
+	delimiter string
+	source    string // the database the dump selects, once seen
+
+	retargeted, definers, collations, sqlModes int
+}
+
+func (f *restoreFilter) run(r io.Reader, w io.Writer) error {
+	br := bufio.NewReaderSize(r, lineBuf)
+	partial := false
+	for {
+		line, err := br.ReadSlice('\n')
+		if len(line) > 0 {
+			out := line
+			if !partial && !errors.Is(err, bufio.ErrBufferFull) {
+				// Long lines are data (extended INSERTs): never rewritten.
+				var ferr error
+				if out, ferr = f.line(line); ferr != nil {
+					return ferr
+				}
+			}
+			if _, werr := w.Write(out); werr != nil {
+				return werr
+			}
+		}
+		switch {
+		case errors.Is(err, bufio.ErrBufferFull):
+			partial = true
+		case err == nil:
+			partial = false
+		case errors.Is(err, io.EOF):
+			return nil
+		default:
+			return err
+		}
+	}
+}
+
+func (f *restoreFilter) line(b []byte) ([]byte, error) {
+	text := strings.TrimSpace(string(b))
+	nl := b[len(strings.TrimRight(string(b), "\r\n")):]
+	if f.OracleClient && sandboxRe.MatchString(text) {
+		return nl, nil
+	}
+	if m := delimiterRe.FindStringSubmatch(text); m != nil {
+		f.delimiter = m[1]
+		return b, nil
+	}
+	if f.Target != "" && f.delimiter == ";" {
+		for _, re := range []*regexp.Regexp{useRe, createDBRe, dropDBRe} {
+			m := re.FindStringSubmatch(text)
+			if m == nil {
+				continue
+			}
+			name := unquoteIdent(m[1])
+			switch {
+			case f.source == "":
+				f.source = name
+			case name != f.source:
+				return nil, fmt.Errorf("the dump holds more than one database (%s, %s): restore it without a database name, or one database at a time", f.source, name)
+			}
+			f.retargeted++
+			return nl, nil
+		}
+	}
+	out := strings.TrimRight(string(b), "\r\n")
+	changed := false
+	if f.NoDefiner && definesObject(text) {
+		if n := len(definerRe.FindAllStringIndex(out, -1)); n > 0 {
+			f.definers += n
+			out, changed = definerRe.ReplaceAllString(out, ""), true
+		}
+	}
+	if f.MySQLServer && !dataRe.MatchString(text) {
+		if n := len(uca1400Re.FindAllStringIndex(out, -1)); n > 0 {
+			f.collations += n
+			out, changed = uca1400Re.ReplaceAllString(out, "utf8mb4_0900_$1"), true
+		}
+		if sqlModeRe.MatchString(out) {
+			fixed := sqlModeRe.ReplaceAllStringFunc(out, func(m string) string {
+				p := sqlModeRe.FindStringSubmatch(m)
+				return p[1] + mysqlModes(p[2]) + p[3]
+			})
+			if fixed != out {
+				f.sqlModes++
+				out, changed = fixed, true
+			}
+		}
+	}
+	if !changed {
+		return b, nil
+	}
+	return append([]byte(out), nl...), nil
+}
+
+// definesObject reports whether a line can carry a DEFINER clause: a CREATE
+// statement, or mysqldump's versioned comments around one.
+func definesObject(s string) bool {
+	if len(s) < 6 {
+		return false
+	}
+	head := strings.ToUpper(s[:6])
+	return (head == "CREATE" || strings.HasPrefix(head, "/*!")) && strings.Contains(strings.ToUpper(s), "DEFINER")
+}
+
+func unquoteIdent(s string) string {
+	if len(s) >= 2 {
+		switch q := s[0]; q {
+		case '`', '\'', '"':
+			if s[len(s)-1] == q {
+				return strings.ReplaceAll(s[1:len(s)-1], string([]byte{q, q}), string(q))
+			}
+		}
+	}
+	return s
+}
+
+// mysql8Modes are the sql_mode values MySQL 8 accepts.
+var mysql8Modes = map[string]bool{
+	"ALLOW_INVALID_DATES": true, "ANSI_QUOTES": true, "ERROR_FOR_DIVISION_BY_ZERO": true, "HIGH_NOT_PRECEDENCE": true,
+	"IGNORE_SPACE": true, "NO_AUTO_VALUE_ON_ZERO": true, "NO_BACKSLASH_ESCAPES": true, "NO_DIR_IN_CREATE": true,
+	"NO_ENGINE_SUBSTITUTION": true, "NO_UNSIGNED_SUBTRACTION": true, "NO_ZERO_DATE": true, "NO_ZERO_IN_DATE": true,
+	"ONLY_FULL_GROUP_BY": true, "PAD_CHAR_TO_FULL_LENGTH": true, "PIPES_AS_CONCAT": true, "REAL_AS_FLOAT": true,
+	"STRICT_ALL_TABLES": true, "STRICT_TRANS_TABLES": true, "TIME_TRUNCATE_FRACTIONAL": true,
+	"ANSI": true, "TRADITIONAL": true,
+}
+
+// mysqlModes keeps the modes of a comma-separated sql_mode MySQL 8 knows.
+func mysqlModes(modes string) string {
+	var kept []string
+	for _, m := range strings.Split(modes, ",") {
+		if m = strings.TrimSpace(m); m != "" && mysql8Modes[strings.ToUpper(m)] {
+			kept = append(kept, m)
+		}
+	}
+	return strings.Join(kept, ",")
+}

internal/mysql/restorefilter_test.go

@@ -0,0 +1,156 @@
+package mysql
+
+import (
+	"io"
+	"strings"
+	"testing"
+	"testing/iotest"
+)
+
+func filter(in, target string, noDefiner bool) (string, error) {
+	return filterFor(in, target, noDefiner, false)
+}
+
+func filterFor(in, target string, noDefiner, oracle bool) (string, error) {
+	return filterWith(in, FilterOptions{Target: target, NoDefiner: noDefiner, OracleClient: oracle})
+}
+
+func filterWith(in string, o FilterOptions) (string, error) {
+	rc := RestoreFilter(iotest.OneByteReader(strings.NewReader(in)), o)
+	defer func() { _ = rc.Close() }()
+	out, err := io.ReadAll(rc)
+	return string(out), err
+}
+
+const sampleDump = "/*!40101 SET NAMES utf8 */;\n" +
+	"-- Current Database: `classicmodels`\n" +
+	"/*!40000 DROP DATABASE IF EXISTS `classicmodels`*/;\n" +
+	"CREATE DATABASE /*!32312 IF NOT EXISTS*/`classicmodels` /*!40100 DEFAULT CHARACTER SET latin1 */;\n" +
+	"\n" +
+	"USE `classicmodels`;\n" +
+	"CREATE TABLE `customers` (`id` int, `note` text);\n" +
+	"INSERT INTO `customers` VALUES (1,'USE `other`;'),(2,'CREATE DEFINER=`a`@`b` VIEW');\n" +
+	"/*!50001 CREATE ALGORITHM=UNDEFINED */\n" +
+	"/*!50013 DEFINER=`u_83c8ac41`@`%` SQL SECURITY DEFINER */\n" +
+	"/*!50001 VIEW `v` AS select 1 */;\n" +
+	"DELIMITER ;;\n" +
+	"CREATE DEFINER=`root`@`localhost` PROCEDURE `p`()\n" +
+	"BEGIN\n" +
+	"  SELECT 1;\n" +
+	"END ;;\n" +
+	"/*!50003 CREATE*/ /*!50017 DEFINER='app'@'%'*/ /*!50003 TRIGGER `t` BEFORE INSERT ON `customers` FOR EACH ROW SET NEW.id = 1 */;;\n" +
+	"DELIMITER ;\n"
+
+func TestRestoreFilterRetargets(t *testing.T) {
+	out, err := filter(sampleDump, "mail", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, gone := range []string{"DROP DATABASE", "CREATE DATABASE", "\nUSE `classicmodels`;"} {
+		if strings.Contains(out, gone) {
+			t.Errorf("%q kept:\n%s", gone, out)
+		}
+	}
+	if !strings.Contains(out, "'USE `other`;'") || !strings.Contains(out, "DEFINER=`root`") {
+		t.Errorf("data or definers changed:\n%s", out)
+	}
+	if strings.Count(out, "\n") != strings.Count(sampleDump, "\n") {
+		t.Errorf("line count changed")
+	}
+
+	// No target (all-in-one): untouched.
+	if out, _ := filter(sampleDump, "", false); out != sampleDump {
+		t.Errorf("all-in-one dump changed:\n%s", out)
+	}
+}
+
+func TestRestoreFilterDefiners(t *testing.T) {
+	out, err := filter(sampleDump, "mail", true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{
+		"/*!50013 SQL SECURITY DEFINER */\n",
+		"CREATE PROCEDURE `p`()\n",
+		"/*!50003 CREATE*/ /*!50017*/ /*!50003 TRIGGER",
+		"(2,'CREATE DEFINER=`a`@`b` VIEW')", // data
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("want %q:\n%s", want, out)
+		}
+	}
+	if strings.Contains(out, "u_83c8ac41") || strings.Contains(out, "`root`@") || strings.Contains(out, "'app'@") {
+		t.Errorf("definer kept:\n%s", out)
+	}
+}
+
+func TestRestoreFilterSeveralDatabases(t *testing.T) {
+	_, err := filter("USE `a`;\nCREATE TABLE t(i int);\nUSE `b`;\n", "mail", false)
+	if err == nil || !strings.Contains(err.Error(), "more than one database (a, b)") {
+		t.Fatalf("err = %v", err)
+	}
+	if _, err := filter("CREATE DATABASE IF NOT EXISTS shop;\nuse shop\n", "mail", false); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCanSetDefiner(t *testing.T) {
+	for grants, want := range map[string]bool{
+		"GRANT USAGE ON *.* TO `u_1`@`%`\nGRANT ALL PRIVILEGES ON `blog`.* TO `u_1`@`%`":      false,
+		"GRANT USAGE ON *.* TO `superman`@`%`":                                                false,
+		"GRANT SELECT, INSERT, SUPER ON *.* TO `root`@`%` WITH GRANT OPTION":                  true,
+		"GRANT USAGE ON *.* TO `a`@`%`\nGRANT SET_USER_ID,XA_RECOVER_ADMIN ON *.* TO `a`@`%`": true,
+		"GRANT ALL PRIVILEGES ON *.* TO `root`@`localhost` WITH GRANT OPTION":                 true,
+		"GRANT SET USER ON *.* TO `m`@`%`":                                                    true,
+	} {
+		if got := canSetDefiner(grants); got != want {
+			t.Errorf("%q: %v", grants, got)
+		}
+	}
+}
+
+func TestRestoreFilterSandbox(t *testing.T) {
+	in := "/*M!999999\\- enable the sandbox mode */ \n-- MariaDB dump 10.20\nCREATE TABLE t (i int);\n"
+	if out, _ := filterFor(in, "mail", false, true); out != "\n-- MariaDB dump 10.20\nCREATE TABLE t (i int);\n" {
+		t.Fatalf("oracle client: %q", out)
+	}
+	if out, _ := filterFor(in, "mail", false, false); out != in {
+		t.Fatalf("mariadb client: %q", out)
+	}
+}
+
+func TestRestoreFilterCollations(t *testing.T) {
+	in := "  `name` varchar(100) COLLATE utf8mb4_uca1400_ai_ci DEFAULT NULL,\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_uca1400_as_cs;\n" +
+		"/*!50003 SET collation_connection  = utf8mb4_uca1400_ai_ci */ ;\n" +
+		"INSERT INTO `t` VALUES ('utf8mb4_uca1400_ai_ci');\n" +
+		"  `x` text COLLATE utf8mb4_uca1400_swedish_ai_ci,\n"
+	want := "  `name` varchar(100) COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs;\n" +
+		"/*!50003 SET collation_connection  = utf8mb4_0900_ai_ci */ ;\n" +
+		"INSERT INTO `t` VALUES ('utf8mb4_uca1400_ai_ci');\n" +
+		"  `x` text COLLATE utf8mb4_uca1400_swedish_ai_ci,\n"
+	if out, _ := filterWith(in, FilterOptions{Target: "mail", MySQLServer: true}); out != want {
+		t.Fatalf("got:\n%s", out)
+	}
+	if out, _ := filterWith(in, FilterOptions{Target: "mail"}); out != in {
+		t.Fatalf("MariaDB target changed:\n%s", out)
+	}
+}
+
+func TestRestoreFilterSQLModes(t *testing.T) {
+	in := "/*!50003 SET sql_mode              = 'STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION' */ ;\n" +
+		"/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO,SIMULTANEOUS_ASSIGNMENT' */;\n" +
+		"/*!50003 SET sql_mode              = @saved_sql_mode */ ;\n" +
+		"INSERT INTO `t` VALUES ('sql_mode=''NO_AUTO_CREATE_USER''');\n"
+	want := "/*!50003 SET sql_mode              = 'STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION' */ ;\n" +
+		"/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;\n" +
+		"/*!50003 SET sql_mode              = @saved_sql_mode */ ;\n" +
+		"INSERT INTO `t` VALUES ('sql_mode=''NO_AUTO_CREATE_USER''');\n"
+	if out, _ := filterWith(in, FilterOptions{Target: "mail", MySQLServer: true}); out != want {
+		t.Fatalf("got:\n%s", out)
+	}
+	if out, _ := filterWith(in, FilterOptions{Target: "mail"}); out != in {
+		t.Fatalf("MariaDB target changed:\n%s", out)
+	}
+}