Merge pull request #30 from xtrimf/support_for_triggers_no_rollback

Support for triggers no rollback

Author: shlomi-noach

go/base/context.go

@@ -14,6 +14,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unicode/utf8"
 
 	uuid "github.com/satori/go.uuid"
 
@@ -225,6 +226,11 @@ type MigrationContext struct {
 	MigrationIterationRangeMaxValues *sql.ColumnValues
 	ForceTmpTableName                string
 
+	IncludeTriggers     bool
+	RemoveTriggerSuffix bool
+	TriggerSuffix       string
+	Triggers            []mysql.Trigger
+
 	recentBinlogCoordinates mysql.BinlogCoordinates
 
 	Log Logger
@@ -854,3 +860,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

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -60,6 +61,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

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"regexp"
 	"syscall"
 
 	"github.com/github/gh-ost/go/base"
@@ -128,6 +129,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")
@@ -236,6 +241,19 @@ func main() {
 	if *replicationLagQuery != "" {
 		migrationContext.Log.Warningf("--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())

go/logic/inspect.go

@@ -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))

go/logic/migrator.go

@@ -615,6 +615,12 @@ func (this *Migrator) cutOverTwoStep() (err error) {
 	if err := this.retryOperation(this.waitForEventsUpToLock); err != nil {
 		return err
 	}
+	// If we need to create triggers we need to do it here (only create part)
+	if this.migrationContext.IncludeTriggers && len(this.migrationContext.Triggers) > 0 {
+		if this.retryOperation(this.applier.CreateTriggersOnGhost); err != nil {
+			return err
+		}
+	}
 	if err := this.retryOperation(this.applier.SwapTablesQuickAndBumpy); err != nil {
 		return err
 	}
@@ -663,6 +669,13 @@ func (this *Migrator) atomicCutOver() (err error) {
 		return this.migrationContext.Log.Errore(err)
 	}
 
+	// If we need to create triggers we need to do it here (only create part)
+	if this.migrationContext.IncludeTriggers && len(this.migrationContext.Triggers) > 0 {
+		if err := this.applier.CreateTriggersOnGhost(); err != nil {
+			this.migrationContext.Log.Errore(err)
+		}
+	}
+
 	// Step 2
 	// We now attempt an atomic RENAME on original & ghost tables, and expect it to block.
 	this.migrationContext.RenameTablesStartTime = time.Now()
@@ -720,6 +733,7 @@ func (this *Migrator) atomicCutOver() (err error) {
 	// ooh nice! We're actually truly and thankfully done
 	lockAndRenameDuration := this.migrationContext.RenameTablesEndTime.Sub(this.migrationContext.LockTablesStartTime)
 	this.migrationContext.Log.Infof("Lock & rename duration: %s. During this time, queries on %s were blocked", lockAndRenameDuration, sql.EscapeName(this.migrationContext.OriginalTableName))
+
 	return nil
 }
 

go/mysql/utils.go

@@ -30,6 +30,13 @@ type ReplicationLagResult struct {
 	Err error
 }
 
+type Trigger struct {
+	Name      string
+	Event     string
+	Statement string
+	Timing    string
+}
+
 func NewNoReplicationLagResult() *ReplicationLagResult {
 	return &ReplicationLagResult{Lag: 0, Err: nil}
 }
@@ -205,3 +212,24 @@ func GetTableColumns(db *gosql.DB, databaseName, tableName string) (*sql.ColumnL
 	}
 	return sql.NewColumnList(columnNames), sql.NewColumnList(virtualColumnNames), nil
 }
+
+// GetTriggers reads trigger list from given table
+func GetTriggers(db *gosql.DB, databaseName, tableName string) (triggers []Trigger, err error) {
+	query := fmt.Sprintf(`select trigger_name as name, event_manipulation as event, action_statement as statement, action_timing as timing
+	from information_schema.triggers 
+	where trigger_schema = '%s' and event_object_table = '%s'`, databaseName, tableName)
+
+	err = sqlutils.QueryRowsMap(db, query, func(rowMap sqlutils.RowMap) error {
+		triggers = append(triggers, Trigger{
+			Name:      rowMap.GetString("name"),
+			Event:     rowMap.GetString("event"),
+			Statement: rowMap.GetString("statement"),
+			Timing:    rowMap.GetString("timing"),
+		})
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return triggers, nil
+}