Only allow -switch-to-rbr when inspecting replica

go/base/context.go

@@ -14,6 +14,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unicode/utf8"
 
 	uuid "github.com/satori/go.uuid"
 
@@ -219,6 +220,7 @@ type MigrationContext struct {
 	GhostTableUniqueKeys             [](*sql.UniqueKey)
 	UniqueKey                        *sql.UniqueKey
 	SharedColumns                    *sql.ColumnList
+	SharedVirtualColumns             *sql.ColumnList
 	ColumnRenameMap                  map[string]string
 	DroppedColumnsMap                map[string]bool
 	MappedSharedColumns              *sql.ColumnList
@@ -229,6 +231,11 @@ type MigrationContext struct {
 	MigrationIterationRangeMaxValues *sql.ColumnValues
 	ForceTmpTableName                string
 
+	IncludeTriggers     bool
+	RemoveTriggerSuffix bool
+	TriggerSuffix       string
+	Triggers            []mysql.Trigger
+
 	recentBinlogCoordinates mysql.BinlogCoordinates
 
 	Log Logger
@@ -888,3 +895,20 @@ func (this *MigrationContext) ReadConfigFile() error {
 
 	return nil
 }
+
+// getGhostTriggerName generates the name of a ghost trigger, based on original trigger name
+// or a given trigger name
+func (this *MigrationContext) GetGhostTriggerName(triggerName string) string {
+	if this.RemoveTriggerSuffix && strings.HasSuffix(triggerName, this.TriggerSuffix) {
+		return strings.TrimSuffix(triggerName, this.TriggerSuffix)
+	}
+	// else
+	return triggerName + this.TriggerSuffix
+}
+
+// validateGhostTriggerLength check if the ghost trigger name length is not more than 64 characters
+func (this *MigrationContext) ValidateGhostTriggerLengthBelowMaxLength(triggerName string) bool {
+	ghostTriggerName := this.GetGhostTriggerName(triggerName)
+
+	return utf8.RuneCountInString(ghostTriggerName) <= mysql.MaxTableNameLength
+}

go/base/context_test.go

@@ -8,6 +8,7 @@ package base
 import (
 	"io/ioutil"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -59,6 +60,68 @@ func TestGetTableNames(t *testing.T) {
 	}
 }
 
+func TestGetTriggerNames(t *testing.T) {
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_gho"
+		test.S(t).ExpectEquals(context.GetGhostTriggerName("my_trigger"), "my_trigger"+context.TriggerSuffix)
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_gho"
+		context.RemoveTriggerSuffix = true
+		test.S(t).ExpectEquals(context.GetGhostTriggerName("my_trigger"), "my_trigger"+context.TriggerSuffix)
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_gho"
+		context.RemoveTriggerSuffix = true
+		test.S(t).ExpectEquals(context.GetGhostTriggerName("my_trigger_gho"), "my_trigger")
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_gho"
+		context.RemoveTriggerSuffix = false
+		test.S(t).ExpectEquals(context.GetGhostTriggerName("my_trigger_gho"), "my_trigger_gho_gho")
+	}
+
+}
+func TestValidateGhostTriggerLengthBelowMaxLength(t *testing.T) {
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_gho"
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength("my_trigger"), true)
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ghost"
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)), false) // 64 characters + "_ghost"
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ghost"
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 3)), true) // 48 characters + "_ghost"
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ghost"
+		context.RemoveTriggerSuffix = true
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)), true) // 64 characters + "_ghost" removed
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ghost"
+		context.RemoveTriggerSuffix = true
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)+"X"), false) // 65 characters + "_ghost" not removed
+	}
+	{
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ghost"
+		context.RemoveTriggerSuffix = true
+		test.S(t).ExpectEquals(context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)+"_ghost"), true) // 70 characters + last "_ghost"  removed
+	}
+}
+
 func TestReadConfigFile(t *testing.T) {
 	{
 		context := NewMigrationContext()

go/cmd/gh-ost/main.go

@@ -11,6 +11,7 @@ import (
 	"net/url"
 	"os"
 	"os/signal"
+	"regexp"
 	"syscall"
 
 	"github.com/github/gh-ost/go/base"
@@ -132,6 +133,10 @@ func main() {
 
 	flag.UintVar(&migrationContext.ReplicaServerId, "replica-server-id", 99999, "server id used by gh-ost process. Default: 99999")
 
+	flag.BoolVar(&migrationContext.IncludeTriggers, "include-triggers", false, "When true, the triggers (if exist) will be created on the new table")
+	flag.StringVar(&migrationContext.TriggerSuffix, "trigger-suffix", "", "Add a suffix to the trigger name (i.e '_v2'). Requires '--include-triggers'")
+	flag.BoolVar(&migrationContext.RemoveTriggerSuffix, "remove-trigger-suffix-if-exists", false, "Remove given suffix from name of trigger. Requires '--include-triggers' and '--trigger-suffix'")
+
 	maxLoad := flag.String("max-load", "", "Comma delimited status-name=threshold. e.g: 'Threads_running=100,Threads_connected=500'. When status exceeds threshold, app throttles writes")
 	criticalLoad := flag.String("critical-load", "", "Comma delimited status-name=threshold, same format as --max-load. When status exceeds threshold, app panics and quits")
 	flag.Int64Var(&migrationContext.CriticalLoadIntervalMilliseconds, "critical-load-interval-millis", 0, "When 0, migration immediately bails out upon meeting critical-load. When non-zero, a second check is done after given interval, and migration only bails out if 2nd check still meets critical load")
@@ -245,6 +250,19 @@ func main() {
 	if *replicationLagQuery != "" {
 		migrationContext.Log.Warning("--replication-lag-query is deprecated")
 	}
+	if migrationContext.IncludeTriggers && migrationContext.TriggerSuffix == "" {
+		migrationContext.Log.Fatalf("--trigger-suffix must be used with --include-triggers")
+	}
+	if !migrationContext.IncludeTriggers && migrationContext.TriggerSuffix != "" {
+		migrationContext.Log.Fatalf("--trigger-suffix cannot be be used without --include-triggers")
+	}
+	if migrationContext.TriggerSuffix != "" {
+		regex := regexp.MustCompile(`^[\da-zA-Z_]+$`)
+
+		if !regex.Match([]byte(migrationContext.TriggerSuffix)) {
+			migrationContext.Log.Fatalf("--trigger-suffix must contain only alpha numeric characters and underscore (0-9,a-z,A-Z,_)")
+		}
+	}
 
 	switch *cutOver {
 	case "atomic", "default", "":

go/logic/applier.go

@@ -334,6 +334,56 @@ func (this *Applier) dropTable(tableName string) error {
 	return nil
 }
 
+// dropTriggers drop the triggers on the applied host
+func (this *Applier) DropTriggersFromGhost() error {
+	if len(this.migrationContext.Triggers) > 0 {
+		for _, trigger := range this.migrationContext.Triggers {
+			triggerName := this.migrationContext.GetGhostTriggerName(trigger.Name)
+			query := fmt.Sprintf("drop trigger if exists %s", sql.EscapeName(triggerName))
+			_, err := sqlutils.ExecNoPrepare(this.db, query)
+			if err != nil {
+				return err
+			}
+			this.migrationContext.Log.Infof("Trigger '%s' dropped", triggerName)
+		}
+	}
+	return nil
+}
+
+// createTriggers creates the triggers on the applied host
+func (this *Applier) createTriggers(tableName string) error {
+	if len(this.migrationContext.Triggers) > 0 {
+		for _, trigger := range this.migrationContext.Triggers {
+			triggerName := this.migrationContext.GetGhostTriggerName(trigger.Name)
+			query := fmt.Sprintf(`create /* gh-ost */ trigger %s %s %s on %s.%s for each row
+		%s`,
+				sql.EscapeName(triggerName),
+				trigger.Timing,
+				trigger.Event,
+				sql.EscapeName(this.migrationContext.DatabaseName),
+				sql.EscapeName(tableName),
+				trigger.Statement,
+			)
+			this.migrationContext.Log.Infof("Createing trigger %s on %s.%s",
+				sql.EscapeName(triggerName),
+				sql.EscapeName(this.migrationContext.DatabaseName),
+				sql.EscapeName(tableName),
+			)
+			if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
+				return err
+			}
+		}
+		this.migrationContext.Log.Infof("Triggers created on %s", tableName)
+	}
+	return nil
+}
+
+// CreateTriggers creates the original triggers on applier host
+func (this *Applier) CreateTriggersOnGhost() error {
+	err := this.createTriggers(this.migrationContext.GetGhostTableName())
+	return err
+}
+
 // DropChangelogTable drops the changelog table on the applier host
 func (this *Applier) DropChangelogTable() error {
 	return this.dropTable(this.migrationContext.GetChangelogTableName())
@@ -1089,7 +1139,7 @@ func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) (result
 				results = append(results, this.buildDMLEventQuery(dmlEvent)...)
 				return results
 			}
-			query, sharedArgs, uniqueKeyArgs, err := sql.BuildDMLUpdateQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, &this.migrationContext.UniqueKey.Columns, dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
+			query, sharedArgs, uniqueKeyArgs, err := sql.BuildDMLUpdateQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, this.migrationContext.SharedVirtualColumns, &this.migrationContext.UniqueKey.Columns, dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
 			args := sqlutils.Args()
 			args = append(args, sharedArgs...)
 			args = append(args, uniqueKeyArgs...)

go/logic/inspect.go

@@ -172,7 +172,7 @@ func (this *Inspector) inspectOriginalAndGhostTables() (err error) {
 		}
 	}
 
-	this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns = this.getSharedColumns(this.migrationContext.OriginalTableColumns, this.migrationContext.GhostTableColumns, this.migrationContext.OriginalTableVirtualColumns, this.migrationContext.GhostTableVirtualColumns, this.migrationContext.ColumnRenameMap)
+	this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, this.migrationContext.SharedVirtualColumns = this.getSharedColumns(this.migrationContext.OriginalTableColumns, this.migrationContext.GhostTableColumns, this.migrationContext.OriginalTableVirtualColumns, this.migrationContext.GhostTableVirtualColumns, this.migrationContext.ColumnRenameMap)
 	this.migrationContext.Log.Infof("Shared columns are %s", this.migrationContext.SharedColumns)
 	// By fact that a non-empty unique key exists we also know the shared columns are non-empty
 
@@ -483,7 +483,7 @@ func (this *Inspector) validateTableForeignKeys(allowChildForeignKeys bool) erro
 	return nil
 }
 
-// validateTableTriggers makes sure no triggers exist on the migrated table
+// validateTableTriggers makes sure no triggers exist on the migrated table. if --include_triggers is used then it fetches the triggers
 func (this *Inspector) validateTableTriggers() error {
 	query := `
 		SELECT COUNT(*) AS num_triggers
@@ -505,12 +505,74 @@ func (this *Inspector) validateTableTriggers() error {
 		return err
 	}
 	if numTriggers > 0 {
-		return this.migrationContext.Log.Errorf("Found triggers on %s.%s. Triggers are not supported at this time. Bailing out", sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
+		if this.migrationContext.IncludeTriggers {
+			this.migrationContext.Log.Infof("Found %d triggers on %s.%s.", numTriggers, sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
+			this.migrationContext.Triggers, err = mysql.GetTriggers(this.db, this.migrationContext.DatabaseName, this.migrationContext.OriginalTableName)
+			if err != nil {
+				return err
+			}
+			if err := this.validateGhostTriggersDontExist(); err != nil {
+				return err
+			}
+			if err := this.validateGhostTriggersLength(); err != nil {
+				return err
+			}
+			return nil
+		}
+		return this.migrationContext.Log.Errorf("Found triggers on %s.%s. Tables with triggers are supported only when using \"include-triggers\" flag. Bailing out", sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
+
 	}
 	this.migrationContext.Log.Debugf("Validated no triggers exist on table")
 	return nil
 }
 
+// verifyTriggersDontExist verifies before createing new triggers we want to make sure these triggers dont exist already in the DB
+func (this *Inspector) validateGhostTriggersDontExist() error {
+	if len(this.migrationContext.Triggers) > 0 {
+		var foundTriggers []string
+		for _, trigger := range this.migrationContext.Triggers {
+			triggerName := this.migrationContext.GetGhostTriggerName(trigger.Name)
+			query := "select 1 from information_schema.triggers where trigger_name = ? and trigger_schema = ?"
+			err := sqlutils.QueryRowsMap(this.db, query, func(rowMap sqlutils.RowMap) error {
+				triggerExists := rowMap.GetInt("1")
+				if triggerExists == 1 {
+					foundTriggers = append(foundTriggers, triggerName)
+				}
+				return nil
+			},
+				triggerName,
+				this.migrationContext.DatabaseName,
+			)
+			if err != nil {
+				return err
+			}
+		}
+		if len(foundTriggers) > 0 {
+			return this.migrationContext.Log.Errorf("Found gh-ost triggers (%s). Please use a different suffix or drop them. Bailing out", strings.Join(foundTriggers, ","))
+		}
+	}
+
+	return nil
+}
+
+func (this *Inspector) validateGhostTriggersLength() error {
+	if len(this.migrationContext.Triggers) > 0 {
+		var foundTriggers []string
+		for _, trigger := range this.migrationContext.Triggers {
+			triggerName := this.migrationContext.GetGhostTriggerName(trigger.Name)
+			if ok := this.migrationContext.ValidateGhostTriggerLengthBelowMaxLength(triggerName); !ok {
+				foundTriggers = append(foundTriggers, triggerName)
+			}
+
+		}
+		if len(foundTriggers) > 0 {
+			return this.migrationContext.Log.Errorf("Gh-ost triggers (%s) length > %d characters. Bailing out", strings.Join(foundTriggers, ","), mysql.MaxTableNameLength)
+		}
+	}
+
+	return nil
+}
+
 // estimateTableRowsViaExplain estimates number of rows on original table
 func (this *Inspector) estimateTableRowsViaExplain() error {
 	query := fmt.Sprintf(`explain select /* gh-ost */ * from %s.%s where 1=1`, sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
@@ -742,8 +804,9 @@ func (this *Inspector) getSharedUniqueKeys(originalUniqueKeys, ghostUniqueKeys [
 }
 
 // getSharedColumns returns the intersection of two lists of columns in same order as the first list
-func (this *Inspector) getSharedColumns(originalColumns, ghostColumns *sql.ColumnList, originalVirtualColumns, ghostVirtualColumns *sql.ColumnList, columnRenameMap map[string]string) (*sql.ColumnList, *sql.ColumnList) {
+func (this *Inspector) getSharedColumns(originalColumns, ghostColumns *sql.ColumnList, originalVirtualColumns, ghostVirtualColumns *sql.ColumnList, columnRenameMap map[string]string) (*sql.ColumnList, *sql.ColumnList, *sql.ColumnList) {
 	sharedColumnNames := []string{}
+	sharedVirtualColumnNames := []string{}
 	for _, originalColumn := range originalColumns.Names() {
 		isSharedColumn := false
 		for _, ghostColumn := range ghostColumns.Names() {
@@ -776,6 +839,29 @@ func (this *Inspector) getSharedColumns(originalColumns, ghostColumns *sql.Colum
 			sharedColumnNames = append(sharedColumnNames, originalColumn)
 		}
 	}
+	// Virtual columns
+	for _, originalColumn := range originalVirtualColumns.Names() {
+		isSharedColumn := false
+		for _, ghostColumn := range ghostVirtualColumns.Names() {
+			if strings.EqualFold(originalColumn, ghostColumn) {
+				isSharedColumn = true
+				break
+			}
+			if strings.EqualFold(columnRenameMap[originalColumn], ghostColumn) {
+				isSharedColumn = true
+				break
+			}
+		}
+		for droppedColumn := range this.migrationContext.DroppedColumnsMap {
+			if strings.EqualFold(originalColumn, droppedColumn) {
+				isSharedColumn = false
+				break
+			}
+		}
+		if isSharedColumn {
+			sharedVirtualColumnNames = append(sharedVirtualColumnNames, originalColumn)
+		}
+	}
 	mappedSharedColumnNames := []string{}
 	for _, columnName := range sharedColumnNames {
 		if mapped, ok := columnRenameMap[columnName]; ok {
@@ -784,7 +870,7 @@ func (this *Inspector) getSharedColumns(originalColumns, ghostColumns *sql.Colum
 			mappedSharedColumnNames = append(mappedSharedColumnNames, columnName)
 		}
 	}
-	return sql.NewColumnList(sharedColumnNames), sql.NewColumnList(mappedSharedColumnNames)
+	return sql.NewColumnList(sharedColumnNames), sql.NewColumnList(mappedSharedColumnNames), sql.NewColumnList(sharedVirtualColumnNames)
 }
 
 // showCreateTable returns the `show create table` statement for given table