Add capability 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")