Add CLI parameters for move-tables feature

go/base/context.go

@@ -271,6 +271,16 @@ type MigrationContext struct {
 	SkipMetadataLockCheck             bool
 	IsOpenMetadataLockInstruments     bool
 
+	// move tables:
+	MoveTables struct {
+		TableNames     []string // List of table names to be moved.
+		TargetHost     string   // Target hostname for the move. This must be a primary/writable host.
+		TargetPort     int      // Target MySQL port for the move.
+		TargetUser     string   // Target username for the move. If not specified, it will default to the source user.
+		TargetPass     string   // Target password for the move. If not specified, it will default to the source password.
+		TargetDatabase string   // Target database name for the move. If not specified, it will default to the source database name.
+	}
+
 	Log Logger
 }
 
@@ -1029,6 +1039,11 @@ func (mctx *MigrationContext) CancelContext() {
 	}
 }
 
+// IsMoveTablesMode returns true if gh-ost should be used for moving tables instead of running a schema migration.
+func (mctx *MigrationContext) IsMoveTablesMode() bool {
+	return len(mctx.MoveTables.TableNames) > 0
+}
+
 // SendWithContext attempts to send a value to a channel, but returns early
 // if the context is cancelled. This prevents goroutine deadlocks when the
 // channel receiver has exited due to an error.

go/cmd/gh-ost/main.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"os/signal"
 	"regexp"
+	"strings"
 	"syscall"
 
 	"github.com/github/gh-ost/go/base"
@@ -164,8 +165,16 @@ func main() {
 	version := flag.Bool("version", false, "Print version & exit")
 	checkFlag := flag.Bool("check-flag", false, "Check if another flag exists/supported. This allows for cross-version scripting. Exits with 0 when all additional provided flags exist, nonzero otherwise. You must provide (dummy) values for flags that require a value. Example: gh-ost --check-flag --cut-over-lock-timeout-seconds --nice-ratio 0")
 	flag.StringVar(&migrationContext.ForceTmpTableName, "force-table-names", "", "table name prefix to be used on the temporary tables")
-	flag.CommandLine.SetOutput(os.Stdout)
 
+	// move tables flags
+	moveTables := flag.String("move-tables", "", "Comma delimited list of tables to move. e.g. 'table1,table2,table3'. This is a special mode that allows you to move tables between database clusters. This mode is mutually exclusive with --alter, --table, --test-on-replica, --migrate-on-replica, --execute-on-replica, and --revert.")
+	flag.StringVar(&migrationContext.MoveTables.TargetHost, "target-host", "", "Target MySQL hostname for --move-tables mode. Must be specified if --move-tables is specified.")
+	flag.IntVar(&migrationContext.MoveTables.TargetPort, "target-port", 3306, "Target MySQL port for --move-tables mode. Defaults to 3306.")
+	flag.StringVar(&migrationContext.MoveTables.TargetUser, "target-user", "", "Target MySQL username for --move-tables mode. If not provided, uses the same user as the source connection")
+	flag.StringVar(&migrationContext.MoveTables.TargetPass, "target-password", "", "Target MySQL password for --move-tables mode. If not provided, uses the same password as the source connection")
+	flag.StringVar(&migrationContext.MoveTables.TargetDatabase, "target-database", "", "Target MySQL database name for --move-tables mode. If not provided, uses the same database name as the source connection")
+
+	flag.CommandLine.SetOutput(os.Stdout)
 	flag.Parse()
 
 	if *checkFlag {
@@ -320,6 +329,33 @@ func main() {
 		migrationContext.Log.Warning("--exact-rowcount with --panic-on-warnings: row counts cannot be exact due to warning detection")
 	}
 
+	if *moveTables != "" {
+		if migrationContext.AlterStatement != "" {
+			log.Fatal("--move-tables is mutually exclusive with --alter")
+		}
+		if migrationContext.OriginalTableName != "" {
+			log.Fatal("--move-tables is mutually exclusive with --table")
+		}
+		if migrationContext.TestOnReplica {
+			log.Fatal("--move-tables is mutually exclusive with --test-on-replica")
+		}
+		if migrationContext.MigrateOnReplica {
+			log.Fatal("--move-tables is mutually exclusive with --migrate-on-replica")
+		}
+		if migrationContext.Revert {
+			log.Fatal("--move-tables is mutually exclusive with --revert")
+		}
+		if migrationContext.MoveTables.TargetHost == "" {
+			log.Fatal("--target-host must be specified when using --move-tables")
+		}
+		migrationContext.MoveTables.TableNames = strings.Split(*moveTables, ",")
+		if len(migrationContext.MoveTables.TableNames) > 1 {
+			// Future version will support moving multiple tables at the same time.
+			// For now, we only support moving a single table at a time.
+			log.Fatal("--move-tables currently supports only a single table")
+		}
+	}
+
 	switch *cutOver {
 	case "atomic", "default", "":
 		migrationContext.CutOverType = base.CutOverAtomic
@@ -379,6 +415,8 @@ func main() {
 	var err error
 	if migrationContext.Revert {
 		err = migrator.Revert()
+	} else if migrationContext.IsMoveTablesMode() {
+		err = migrator.MoveTables()
 	} else {
 		err = migrator.Migrate()
 	}

go/logic/migrator.go

@@ -769,6 +769,109 @@ func (mgtr *Migrator) Revert() error {
 	return nil
 }
 
+func (mgtr *Migrator) MoveTables() (err error) {
+	mgtr.migrationContext.Log.Infof("Moving tables %v from %s to %s (%s)",
+		mgtr.migrationContext.MoveTables.TableNames,
+		sql.EscapeName(mgtr.migrationContext.DatabaseName),
+		sql.EscapeName(mgtr.migrationContext.MoveTables.TargetDatabase), mgtr.migrationContext.MoveTables.TargetHost)
+	mgtr.migrationContext.StartTime = time.Now()
+
+	// Ensure context is cancelled on exit (cleanup)
+	defer mgtr.migrationContext.CancelContext()
+
+	if mgtr.migrationContext.Hostname, err = os.Hostname(); err != nil {
+		return err
+	}
+
+	go mgtr.listenOnPanicAbort()
+
+	// Run on-startup hook:
+	if err := mgtr.hooksExecutor.OnStartup(); err != nil {
+		return err
+	}
+
+	// After this point, we'll need to teardown anything that's been started
+	// so we don't leave things hanging around
+	defer mgtr.teardown()
+
+	if err := mgtr.initiateInspector(); err != nil {
+		return err
+	}
+	if err := mgtr.checkAbort(); err != nil {
+		return err
+	}
+	if err := mgtr.initiateApplier(); err != nil {
+		return err
+	}
+	if err := mgtr.checkAbort(); err != nil {
+		return err
+	}
+
+	// Validation complete! Run on-validated hook.
+	if err := mgtr.hooksExecutor.OnValidated(); err != nil {
+		return err
+	}
+
+	if err := mgtr.initiateServer(); err != nil {
+		return err
+	}
+	defer mgtr.server.RemoveSocketFile()
+
+	if err := mgtr.countTableRows(); err != nil {
+		return err
+	}
+	if err := mgtr.addDMLEventsListener(); err != nil {
+		return err
+	}
+	if err := mgtr.applier.ReadMigrationRangeValues(); err != nil {
+		return err
+	}
+
+	mgtr.initiateThrottler()
+
+	// Run on-before-row-copy hook
+	if err := mgtr.hooksExecutor.OnBeforeRowCopy(); err != nil {
+		return err
+	}
+	go func() {
+		if err := mgtr.executeWriteFuncs(); err != nil {
+			// Send error to PanicAbort to trigger abort
+			_ = base.SendWithContext(mgtr.migrationContext.GetContext(), mgtr.migrationContext.PanicAbort, err)
+		}
+	}()
+	go mgtr.iterateChunks()
+	mgtr.migrationContext.MarkRowCopyStartTime()
+	go mgtr.initiateStatus()
+
+	mgtr.migrationContext.Log.Debugf("Operating until row copy is complete")
+	mgtr.consumeRowCopyComplete()
+	mgtr.migrationContext.Log.Infof("Row copy complete")
+	// Check if row copy was aborted due to error
+	if err := mgtr.checkAbort(); err != nil {
+		return err
+	}
+	if err := mgtr.hooksExecutor.OnRowCopyComplete(); err != nil {
+		return err
+	}
+
+	//TODO: cutover here
+
+	if err := mgtr.finalCleanup(); err != nil {
+		return nil
+	}
+	if err := mgtr.hooksExecutor.OnSuccess(false); err != nil {
+		return err
+	}
+	mgtr.migrationContext.Log.Infof("Done moving tables %v from %s to %s (%s)",
+		mgtr.migrationContext.MoveTables.TableNames, sql.EscapeName(mgtr.migrationContext.DatabaseName),
+		sql.EscapeName(mgtr.migrationContext.MoveTables.TargetDatabase), mgtr.migrationContext.MoveTables.TargetHost)
+	// Final check for abort before declaring success
+	if err := mgtr.checkAbort(); err != nil {
+		return err
+	}
+	return nil
+}
+
 // ExecOnFailureHook executes the onFailure hook, and this method is provided as the only external
 // hook access point
 func (mgtr *Migrator) ExecOnFailureHook() (err error) {