Skip to content

Commit 5fe0cb3

Browse files
eldadtseldadts
committed
Trigger support - Not working
1 parent 06fd4d5 commit 5fe0cb3

File tree

6 files changed

+200
-2
lines changed

6 files changed

+200
-2
lines changed

‎go/base/context.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"github.com/github/gh-ost/go/mysql"
2121
"github.com/github/gh-ost/go/sql"
2222
"github.com/openark/golib/log"
23+
"github.com/openark/golib/sqlutils"
2324

2425
"github.com/go-ini/ini"
2526
)
@@ -153,9 +154,10 @@ type MigrationContext struct {
153154
InitiallyDropOldTable bool
154155
InitiallyDropGhostTable bool
155156
TimestampOldTable bool // Should old table name include a timestamp
157+
IncludeTriggers bool
156158
CutOverType CutOver
157159
ReplicaServerId uint
158-
160+
159161
Hostname string
160162
AssumeMasterHostname string
161163
ApplierTimeZone string
@@ -219,6 +221,7 @@ type MigrationContext struct {
219221
MappedSharedColumns *sql.ColumnList
220222
MigrationRangeMinValues *sql.ColumnValues
221223
MigrationRangeMaxValues *sql.ColumnValues
224+
Triggers []*sqlutils.RowMap
222225
Iteration int64
223226
MigrationIterationRangeMinValues *sql.ColumnValues
224227
MigrationIterationRangeMaxValues *sql.ColumnValues

‎go/cmd/gh-ost/main.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,8 @@ func main() {
127127

128128
flag.UintVar(&migrationContext.ReplicaServerId, "replica-server-id", 99999, "server id used by gh-ost process. Default: 99999")
129129

130+
flag.BoolVar(&migrationContext.IncludeTriggers, "include-triggers", false, "When true, the triggers (if exist) will be created on the new table")
131+
130132
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")
131133
criticalLoad := flag.String("critical-load", "", "Comma delimited status-name=threshold, same format as --max-load. When status exceeds threshold, app panics and quits")
132134
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")

‎go/logic/applier.go‎

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -272,6 +272,148 @@ func (this *Applier) dropTable(tableName string) error {
272272
return nil
273273
}
274274

275+
// dropTriggers drop the triggers on the applied host
276+
func (this *Applier) dropTriggers() error {
277+
if len(this.migrationContext.Triggers) > 0 {
278+
for _, trigger := range this.migrationContext.Triggers {
279+
name := (*trigger).GetString("name")
280+
query := fmt.Sprintf("drop trigger if exists %s", sql.EscapeName(name))
281+
if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
282+
return err
283+
}
284+
this.migrationContext.Log.Infof("Trigger '%s' dropped", name)
285+
}
286+
}
287+
return nil
288+
}
289+
290+
// createTriggers creates the triggers on the applied host
291+
func (this *Applier) createTriggers(tableName, suffix string) error {
292+
if len(this.migrationContext.Triggers) > 0 {
293+
for _, trigger := range this.migrationContext.Triggers {
294+
query := fmt.Sprintf(`create /* gh-ost */ trigger %s %s %s on %s.%s for each row
295+
%s`,
296+
sql.EscapeName((*trigger).GetString("name")+suffix),
297+
(*trigger).GetString("timing"),
298+
(*trigger).GetString("event"),
299+
sql.EscapeName(this.migrationContext.DatabaseName),
300+
sql.EscapeName(tableName),
301+
(*trigger).GetString("statement"),
302+
)
303+
this.migrationContext.Log.Infof("Createing trigger %s on %s.%s",
304+
sql.EscapeName((*trigger).GetString("name")+suffix),
305+
sql.EscapeName(this.migrationContext.DatabaseName),
306+
sql.EscapeName(tableName),
307+
)
308+
if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
309+
// tx.Rollback()
310+
return err
311+
}
312+
}
313+
this.migrationContext.Log.Infof("Triggers created on %s", tableName)
314+
}
315+
return nil
316+
}
317+
318+
// createTriggersWithLock creates the triggers on the applied host while locking the table
319+
func (this *Applier) createTriggersWithLock(tableName string) error {
320+
if len(this.migrationContext.Triggers) > 0 {
321+
for _, trigger := range this.migrationContext.Triggers {
322+
triggerName := (*trigger).GetString("name")
323+
this.migrationContext.Log.Infof("Createing trigger %s on %s.%s",
324+
sql.EscapeName(triggerName),
325+
sql.EscapeName(this.migrationContext.DatabaseName),
326+
sql.EscapeName(tableName),
327+
)
328+
329+
// if err := this.LockOriginalTable(); err != nil {
330+
// return err
331+
// }
332+
333+
// tx, err := this.singletonDB.Begin()
334+
tx, err := this.db.Begin()
335+
if err != nil {
336+
return err
337+
}
338+
339+
// if _, err := tx.Exec("DELIMITER ;;"); err != nil {
340+
// tx.Rollback()
341+
// return err
342+
// }
343+
lockQuery := fmt.Sprintf(`lock /* gh-ost */ tables %s.%s write`,
344+
sql.EscapeName(this.migrationContext.DatabaseName),
345+
sql.EscapeName(this.migrationContext.OriginalTableName),
346+
)
347+
this.migrationContext.Log.Infof("Locking %s.%s",
348+
sql.EscapeName(this.migrationContext.DatabaseName),
349+
sql.EscapeName(this.migrationContext.OriginalTableName),
350+
)
351+
if _, err := tx.Exec(lockQuery); err != nil {
352+
return err
353+
}
354+
355+
if _, err := tx.Exec(fmt.Sprintf("drop trigger if exists %s", sql.EscapeName(triggerName+"_gho"))); err != nil {
356+
return err
357+
}
358+
359+
// if _, err := sqlutils.ExecNoPrepare(this.singletonDB, "DELIMITER ;;"); err != nil {
360+
// return err
361+
// }
362+
// if _, err := sqlutils.ExecNoPrepare(this.singletonDB, fmt.Sprintf("drop trigger if exists %s", sql.EscapeName(triggerName+"_gho"))); err != nil {
363+
// return err
364+
// }
365+
366+
query := fmt.Sprintf(`create /* gh-ost */ trigger %s %s %s on %s for each row
367+
%s`,
368+
sql.EscapeName(triggerName),
369+
(*trigger).GetString("timing"),
370+
(*trigger).GetString("event"),
371+
sql.EscapeName(tableName),
372+
(*trigger).GetString("statement"),
373+
)
374+
// spew.Dump(query)
375+
// if _, err := sqlutils.ExecNoPrepare(this.singletonDB, query); err != nil {
376+
// return err
377+
// }
378+
379+
if _, err := tx.Exec(query); err != nil {
380+
tx.Rollback()
381+
return err
382+
}
383+
384+
// if _, err := tx.Exec("delimiter ;"); err != nil {
385+
// tx.Rollback()
386+
// return err
387+
// }
388+
389+
if err := this.UnlockTables(); err != nil {
390+
// tx.Rollback()
391+
return err
392+
}
393+
}
394+
this.migrationContext.Log.Infof("Triggers created on %s", tableName)
395+
}
396+
return nil
397+
}
398+
399+
// ReCreateTriggers creates the original triggers on applier host
400+
func (this *Applier) CreateTriggersOnGhost(tableLocked chan<- error) error {
401+
err := this.createTriggers(this.migrationContext.GetGhostTableName(), "_gho")
402+
tableLocked <- err
403+
return err
404+
}
405+
406+
// ReCreateTriggers creates the original triggers on applier host
407+
func (this *Applier) CreateTriggersOnSource(triggersMigrated chan<- error) error {
408+
if err := this.dropTriggers(); err != nil { // first lets drop original triggers
409+
triggersMigrated <- err
410+
return err
411+
}
412+
err := this.createTriggersWithLock(this.migrationContext.DatabaseName + "." + this.migrationContext.OriginalTableName)
413+
triggersMigrated <- err
414+
return err
415+
}
416+
275417
// DropChangelogTable drops the changelog table on the applier host
276418
func (this *Applier) DropChangelogTable() error {
277419
return this.dropTable(this.migrationContext.GetChangelogTableName())

‎go/logic/inspect.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -505,7 +505,16 @@ func (this *Inspector) validateTableTriggers() error {
505505
return err
506506
}
507507
if numTriggers > 0 {
508-
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))
508+
if this.migrationContext.IncludeTriggers {
509+
this.migrationContext.Log.Infof("Found %d triggers on %s.%s.", numTriggers, sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
510+
this.migrationContext.Triggers, err = mysql.GetTriggers(this.db, this.migrationContext.DatabaseName, this.migrationContext.OriginalTableName)
511+
if err != nil {
512+
return err
513+
}
514+
return nil
515+
}
516+
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))
517+
509518
}
510519
this.migrationContext.Log.Debugf("Validated no triggers exist on table")
511520
return nil

‎go/logic/migrator.go‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -663,6 +663,19 @@ func (this *Migrator) atomicCutOver() (err error) {
663663
return this.migrationContext.Log.Errore(err)
664664
}
665665

666+
// If we need to create triggers we need to do it here (only create part)
667+
if this.migrationContext.IncludeTriggers && len(this.migrationContext.Triggers) > 0 {
668+
triggersCreated := make(chan error, 2)
669+
go func() {
670+
if err := this.applier.CreateTriggersOnGhost(triggersCreated); err != nil {
671+
this.migrationContext.Log.Errore(err)
672+
}
673+
}()
674+
if err := <-triggersCreated; err != nil {
675+
return this.migrationContext.Log.Errore(err)
676+
}
677+
}
678+
666679
// Step 2
667680
// We now attempt an atomic RENAME on original & ghost tables, and expect it to block.
668681
this.migrationContext.RenameTablesStartTime = time.Now()
@@ -720,6 +733,19 @@ func (this *Migrator) atomicCutOver() (err error) {
720733
// ooh nice! We're actually truly and thankfully done
721734
lockAndRenameDuration := this.migrationContext.RenameTablesEndTime.Sub(this.migrationContext.LockTablesStartTime)
722735
this.migrationContext.Log.Infof("Lock & rename duration: %s. During this time, queries on %s were blocked", lockAndRenameDuration, sql.EscapeName(this.migrationContext.OriginalTableName))
736+
737+
// if we completly migrate triggers
738+
if this.migrationContext.IncludeTriggers && len(this.migrationContext.Triggers) > 0 {
739+
triggersMigrated := make(chan error, 2)
740+
go func() {
741+
if err := this.applier.CreateTriggersOnSource(triggersMigrated); err != nil {
742+
this.migrationContext.Log.Errore(err)
743+
}
744+
}()
745+
if err := <-triggersMigrated; err == nil {
746+
return this.migrationContext.Log.Errore(err)
747+
}
748+
}
723749
return nil
724750
}
725751

‎go/mysql/utils.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,3 +205,19 @@ func GetTableColumns(db *gosql.DB, databaseName, tableName string) (*sql.ColumnL
205205
}
206206
return sql.NewColumnList(columnNames), sql.NewColumnList(virtualColumnNames), nil
207207
}
208+
209+
// GetTriggers reads trigger list from given table
210+
func GetTriggers(db *gosql.DB, databaseName, tableName string) (triggers []*sqlutils.RowMap, err error) {
211+
query := fmt.Sprintf(`select trigger_name as name, event_manipulation as event, action_statement as statement, action_timing as timing
212+
from information_schema.triggers
213+
where trigger_schema = '%s' and event_object_table = '%s'`, databaseName, tableName)
214+
215+
err = sqlutils.QueryRowsMap(db, query, func(rowMap sqlutils.RowMap) error {
216+
triggers = append(triggers, &rowMap)
217+
return nil
218+
})
219+
if err != nil {
220+
return nil, err
221+
}
222+
return triggers, nil
223+
}

0 commit comments

Comments
 (0)