Add capability to create consistent transactions

go/mysql/flavor.go

@@ -45,6 +45,10 @@ type flavor interface {
 	// startSlave returns the command to start the slave.
 	startSlaveCommand() string
 
+	// startSlaveUntilAfter will restart replication, but only allow it
+	// to run until `pos` is reached. After reaching pos, replication will be stopped again
+	startSlaveUntilAfter(pos Position) string
+
 	// stopSlave returns the command to stop the slave.
 	stopSlaveCommand() string
 
@@ -146,6 +150,11 @@ func (c *Conn) StartSlaveCommand() string {
 	return c.flavor.startSlaveCommand()
 }
 
+// StartSlaveUntilAfterCommand returns the command to start the slave.
+func (c *Conn) StartSlaveUntilAfterCommand(pos Position) string {
+	return c.flavor.startSlaveUntilAfter(pos)
+}
+
 // StopSlaveCommand returns the command to stop the slave.
 func (c *Conn) StopSlaveCommand() string {
 	return c.flavor.stopSlaveCommand()

go/mysql/flavor_mariadb.go

@@ -41,6 +41,10 @@ func (mariadbFlavor) masterGTIDSet(c *Conn) (GTIDSet, error) {
 	return parseMariadbGTIDSet(qr.Rows[0][0].ToString())
 }
 
+func (mariadbFlavor) startSlaveUntilAfter(pos Position) string {
+	return fmt.Sprintf("START SLAVE UNTIL master_gtid_pos = \"%s\"", pos)
+}
+
 func (mariadbFlavor) startSlaveCommand() string {
 	return "START SLAVE"
 }

go/mysql/flavor_mysql.go

@@ -43,6 +43,10 @@ func (mysqlFlavor) startSlaveCommand() string {
 	return "START SLAVE"
 }
 
+func (mysqlFlavor) startSlaveUntilAfter(pos Position) string {
+	return fmt.Sprintf("START SLAVE UNTIL SQL_AFTER_GTIDS = '%s'", pos)
+}
+
 func (mysqlFlavor) stopSlaveCommand() string {
 	return "STOP SLAVE"
 }

go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go

@@ -80,6 +80,9 @@ type FakeMysqlDaemon struct {
 	// If it doesn't match, SetSlavePosition will return an error.
 	SetSlavePositionPos mysql.Position
 
+	// StartSlaveUntilAfterPos is matched against the input
+	StartSlaveUntilAfterPos mysql.Position
+
 	// SetMasterInput is matched against the input of SetMaster
 	// (as "%v:%v"). If it doesn't match, SetMaster will return an error.
 	SetMasterInput string
@@ -240,6 +243,17 @@ func (fmd *FakeMysqlDaemon) StartSlave(hookExtraEnv map[string]string) error {
 	})
 }
 
+// StartSlaveUntilAfter is part of the MysqlDaemon interface.
+func (fmd *FakeMysqlDaemon) StartSlaveUntilAfter(ctx context.Context, pos mysql.Position) error {
+	if !reflect.DeepEqual(fmd.StartSlaveUntilAfterPos, pos) {
+		return fmt.Errorf("wrong pos for StartSlaveUntilAfter: expected %v got %v", fmd.SetSlavePositionPos, pos)
+	}
+
+	return fmd.ExecuteSuperQueryList(context.Background(), []string{
+		"START SLAVE UNTIL AFTER",
+	})
+}
+
 // StopSlave is part of the MysqlDaemon interface.
 func (fmd *FakeMysqlDaemon) StopSlave(hookExtraEnv map[string]string) error {
 	return fmd.ExecuteSuperQueryList(context.Background(), []string{

go/vt/mysqlctl/mysql_daemon.go

@@ -41,6 +41,7 @@ type MysqlDaemon interface {
 
 	// replication related methods
 	StartSlave(hookExtraEnv map[string]string) error
+	StartSlaveUntilAfter(ctx context.Context, pos mysql.Position) error
 	StopSlave(hookExtraEnv map[string]string) error
 	SlaveStatus() (mysql.SlaveStatus, error)
 	SetSemiSyncEnabled(master, slave bool) error

go/vt/mysqlctl/replication.go

@@ -83,6 +83,19 @@ func (mysqld *Mysqld) StartSlave(hookExtraEnv map[string]string) error {
 	return h.ExecuteOptional()
 }
 
+// StartSlaveUntilAfter starts a slave until replication has come to `targetPos`, then it stops replication
+func (mysqld *Mysqld) StartSlaveUntilAfter(ctx context.Context, targetPos mysql.Position) error {
+	conn, err := getPoolReconnect(ctx, mysqld.dbaPool)
+	if err != nil {
+		return err
+	}
+	defer conn.Recycle()
+
+	queries := []string{conn.StartSlaveUntilAfterCommand(targetPos)}
+
+	return mysqld.executeSuperQueryListConn(ctx, conn, queries)
+}
+
 // StopSlave stops a slave.
 func (mysqld *Mysqld) StopSlave(hookExtraEnv map[string]string) error {
 	h := hook.NewSimpleHook("preflight_stop_slave")