store min and max range values in checkpoint

Author: meiji163

go/base/context.go

@@ -195,7 +195,6 @@ type MigrationContext struct {
 	pointOfInterestTimeMutex               *sync.Mutex
 	lastHeartbeatOnChangelogTime           time.Time
 	lastHeartbeatOnChangelogMutex          *sync.Mutex
-	LastLockProcessedCoords                mysql.BinlogCoordinates
 	CurrentLag                             int64
 	currentProgress                        uint64
 	etaNanoseonds                          int64

go/cmd/gh-ost/main.go

@@ -145,6 +145,7 @@ func main() {
 	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'")
 	flag.BoolVar(&migrationContext.SkipPortValidation, "skip-port-validation", false, "Skip port validation for MySQL connections")
+	flag.BoolVar(&migrationContext.Checkpoint, "checkpoint", false, "Use migration checkpoints")
 
 	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")

go/logic/applier.go

@@ -25,6 +25,7 @@ import (
 	"github.com/github/gh-ost/go/mysql"
 	drivermysql "github.com/go-sql-driver/mysql"
 	"github.com/openark/golib/sqlutils"
+	"sync"
 )
 
 const (
@@ -70,6 +71,13 @@ type Applier struct {
 	finishedMigrating int64
 	name              string
 
+	CurrentCoordinatesMutex sync.Mutex
+	CurrentCoordinates      mysql.BinlogCoordinates
+
+	LastIterationRangeMutex     sync.Mutex
+	LastIterationRangeMinValues *sql.ColumnValues
+	LastIterationRangeMaxValues *sql.ColumnValues
+
 	dmlDeleteQueryBuilder        *sql.DMLDeleteQueryBuilder
 	dmlInsertQueryBuilder        *sql.DMLInsertQueryBuilder
 	dmlUpdateQueryBuilder        *sql.DMLUpdateQueryBuilder
@@ -415,6 +423,8 @@ func (this *Applier) CreateChangelogTable() error {
 }
 
 // Create the checkpoint table to store the chunk copy and applier state.
+// There are two sets of columns with the same types as the shared unique key,
+// one for IterationMinValues and one for IterationMaxValues.
 func (this *Applier) CreateCheckpointTable() error {
 	if err := this.DropCheckpointTable(); err != nil {
 		return err
@@ -428,12 +438,21 @@ func (this *Applier) CreateCheckpointTable() error {
 		if col.MySQLType == "" {
 			return fmt.Errorf("CreateCheckpoinTable: column %s has no type information. applyColumnTypes must be called", sql.EscapeName(col.Name))
 		}
-		colDef := fmt.Sprintf("%s %s", sql.EscapeName(col.Name), col.MySQLType)
+		colDef := fmt.Sprintf("%s %s", sql.EscapeName(col.Name+"_min"), col.MySQLType)
+		if !col.Nullable {
+			colDef += " NOT NULL"
+		}
+		colDefs = append(colDefs, colDef)
+	}
+
+	for _, col := range this.migrationContext.UniqueKey.Columns.Columns() {
+		colDef := fmt.Sprintf("%s %s", sql.EscapeName(col.Name+"_max"), col.MySQLType)
 		if !col.Nullable {
 			colDef += " NOT NULL"
 		}
 		colDefs = append(colDefs, colDef)
 	}
+
 	query := fmt.Sprintf("create /* gh-ost */ table %s.%s (\n %s\n)",
 		sql.EscapeName(this.migrationContext.DatabaseName),
 		sql.EscapeName(this.migrationContext.GetCheckpointTableName()),
@@ -597,6 +616,7 @@ func (this *Applier) WriteChangelogState(value string) (string, error) {
 func (this *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) {
 	var insertId int64
 	uniqueKeyArgs := sqlutils.Args(chk.IterationRangeMin.AbstractValues()...)
+	uniqueKeyArgs = append(uniqueKeyArgs, chk.IterationRangeMax.AbstractValues()...)
 	query, uniqueKeyArgs, err := this.checkpointInsertQueryBuilder.BuildQuery(uniqueKeyArgs)
 	if err != nil {
 		return insertId, err
@@ -611,11 +631,12 @@ func (this *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) {
 }
 
 func (this *Applier) ReadLastCheckpoint(chk *Checkpoint) error {
-	row := this.db.QueryRow(fmt.Sprintf(`select /* gh-ost */ * from %s.%s order by id desc limit 1`, this.migrationContext.DatabaseName, this.migrationContext.GetCheckpointTableName()))
+	row := this.db.QueryRow(fmt.Sprintf(`select /* gh-ost */ * from %s.%s order by gh_ost_chk_id desc limit 1`, this.migrationContext.DatabaseName, this.migrationContext.GetCheckpointTableName()))
 
 	var coordStr string
 	ptrs := []interface{}{&chk.Id, &coordStr, &chk.Iteration}
 	ptrs = append(ptrs, chk.IterationRangeMin.ValuesPointers...)
+	ptrs = append(ptrs, chk.IterationRangeMax.ValuesPointers...)
 	err := row.Scan(ptrs...)
 	if err != nil {
 		if errors.Is(err, gosql.ErrNoRows) {
@@ -777,6 +798,13 @@ func (this *Applier) ReadMigrationRangeValues() error {
 // no further chunk to work through, i.e. we're past the last chunk and are done with
 // iterating the range (and thus done with copying row chunks)
 func (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, err error) {
+	this.LastIterationRangeMutex.Lock()
+	if this.migrationContext.MigrationIterationRangeMinValues != nil && this.migrationContext.MigrationIterationRangeMaxValues != nil {
+		this.LastIterationRangeMinValues = this.migrationContext.MigrationIterationRangeMinValues.Clone()
+		this.LastIterationRangeMaxValues = this.migrationContext.MigrationIterationRangeMaxValues.Clone()
+	}
+	this.LastIterationRangeMutex.Unlock()
+
 	this.migrationContext.MigrationIterationRangeMinValues = this.migrationContext.MigrationIterationRangeMaxValues
 	if this.migrationContext.MigrationIterationRangeMinValues == nil {
 		this.migrationContext.MigrationIterationRangeMinValues = this.migrationContext.MigrationRangeMinValues

go/logic/applier_test.go

@@ -689,7 +689,7 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 	suite.Require().NoError(err)
 
 	// checkpoint table is empty
-	gotChk := &Checkpoint{IterationRangeMin: sql.NewColumnValues(2)}
+	gotChk := &Checkpoint{IterationRangeMin: sql.NewColumnValues(2), IterationRangeMax: sql.NewColumnValues(2)}
 	err = applier.ReadLastCheckpoint(gotChk)
 	suite.Require().ErrorIs(err, NoCheckpointFoundError)
 
@@ -699,7 +699,8 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 
 	chk := &Checkpoint{
 		LastTrxCoords:     coords,
-		IterationRangeMin: applier.migrationContext.MigrationRangeMaxValues,
+		IterationRangeMin: applier.migrationContext.MigrationRangeMinValues,
+		IterationRangeMax: applier.migrationContext.MigrationRangeMaxValues,
 		Iteration:         2,
 	}
 	id, err := applier.WriteCheckpoint(chk)
@@ -712,6 +713,7 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 	suite.Require().Equal(chk.Iteration, gotChk.Iteration)
 	suite.Require().Equal(chk.LastTrxCoords.String(), gotChk.LastTrxCoords.String())
 	suite.Require().Equal(chk.IterationRangeMin.String(), gotChk.IterationRangeMin.String())
+	suite.Require().Equal(chk.IterationRangeMax.String(), gotChk.IterationRangeMax.String())
 }
 
 func TestApplier(t *testing.T) {

go/logic/migrator.go

@@ -26,6 +26,8 @@ var (
 	ErrMigratorUnsupportedRenameAlter = errors.New("ALTER statement seems to RENAME the table. This is not supported, and you should run your RENAME outside gh-ost.")
 	ErrMigrationNotAllowedOnMaster    = errors.New("It seems like this migration attempt to run directly on master. Preferably it would be executed on a replica (this reduces load from the master). To proceed please provide --allow-on-master.")
 	RetrySleepFn                      = time.Sleep
+	checkpointInterval                = 10 * time.Second // 5 * time.Minute
+	checkpointTimeout                 = 2 * time.Second
 )
 
 type ChangelogState string
@@ -46,15 +48,16 @@ type tableWriteFunc func() error
 type applyEventStruct struct {
 	writeFunc *tableWriteFunc
 	dmlEvent  *binlog.BinlogDMLEvent
+	coords    mysql.BinlogCoordinates
 }
 
 func newApplyEventStructByFunc(writeFunc *tableWriteFunc) *applyEventStruct {
 	result := &applyEventStruct{writeFunc: writeFunc}
 	return result
 }
 
-func newApplyEventStructByDML(dmlEvent *binlog.BinlogDMLEvent) *applyEventStruct {
-	result := &applyEventStruct{dmlEvent: dmlEvent}
+func newApplyEventStructByDML(dmlEntry *binlog.BinlogEntry) *applyEventStruct {
+	result := &applyEventStruct{dmlEvent: dmlEntry.DmlEvent, coords: dmlEntry.Coordinates}
 	return result
 }
 
@@ -431,6 +434,9 @@ func (this *Migrator) Migrate() (err error) {
 	go this.iterateChunks()
 	this.migrationContext.MarkRowCopyStartTime()
 	go this.initiateStatus()
+	if this.migrationContext.Checkpoint {
+		go this.checkpointLoop()
+	}
 
 	this.migrationContext.Log.Debugf("Operating until row copy is complete")
 	this.consumeRowCopyComplete()
@@ -1086,7 +1092,7 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 		atomic.LoadInt64(&this.migrationContext.TotalDMLEventsApplied),
 		len(this.applyEventsQueue), cap(this.applyEventsQueue),
 		base.PrettifyDurationOutput(elapsedTime), base.PrettifyDurationOutput(this.migrationContext.ElapsedRowCopyTime()),
-		currentBinlogCoordinates,
+		currentBinlogCoordinates.DisplayString(),
 		this.migrationContext.GetCurrentLagDuration().Seconds(),
 		this.migrationContext.TimeSinceLastHeartbeatOnChangelog().Seconds(),
 		state,
@@ -1123,8 +1129,8 @@ func (this *Migrator) initiateStreaming() error {
 		false,
 		this.migrationContext.DatabaseName,
 		this.migrationContext.GetChangelogTableName(),
-		func(dmlEvent *binlog.BinlogDMLEvent) error {
-			return this.onChangelogEvent(dmlEvent)
+		func(dmlEntry *binlog.BinlogEntry) error {
+			return this.onChangelogEvent(dmlEntry.DmlEvent)
 		},
 	)
 
@@ -1157,8 +1163,8 @@ func (this *Migrator) addDMLEventsListener() error {
 		false,
 		this.migrationContext.DatabaseName,
 		this.migrationContext.OriginalTableName,
-		func(dmlEvent *binlog.BinlogDMLEvent) error {
-			this.applyEventsQueue <- newApplyEventStructByDML(dmlEvent)
+		func(dmlEntry *binlog.BinlogEntry) error {
+			this.applyEventsQueue <- newApplyEventStructByDML(dmlEntry)
 			return nil
 		},
 	)
@@ -1342,6 +1348,11 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 		if err := this.retryOperation(applyEventFunc); err != nil {
 			return this.migrationContext.Log.Errore(err)
 		}
+		// update applier coordinates
+		this.applier.CurrentCoordinatesMutex.Lock()
+		this.applier.CurrentCoordinates = eventStruct.coords
+		this.applier.CurrentCoordinatesMutex.Unlock()
+
 		if nonDmlStructToApply != nil {
 			// We pulled DML events from the queue, and then we hit a non-DML event. Wait!
 			// We need to handle it!
@@ -1353,6 +1364,65 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 	return nil
 }
 
+func (this *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) {
+	coords := this.eventsStreamer.GetCurrentBinlogCoordinates()
+	this.applier.LastIterationRangeMutex.Lock()
+	if this.applier.LastIterationRangeMaxValues == nil || this.applier.LastIterationRangeMinValues == nil {
+		this.applier.LastIterationRangeMutex.Unlock()
+		return nil, errors.New("iteration range is empty, not checkpointing...")
+	}
+	chk := &Checkpoint{
+		IterationRangeMin: this.applier.LastIterationRangeMinValues.Clone(),
+		IterationRangeMax: this.applier.LastIterationRangeMaxValues.Clone(),
+		LastTrxCoords:     coords,
+	}
+	this.applier.LastIterationRangeMutex.Unlock()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			this.applier.CurrentCoordinatesMutex.Lock()
+			if coords.SmallerThanOrEquals(this.applier.CurrentCoordinates) {
+				id, err := this.applier.WriteCheckpoint(chk)
+				chk.Id = id
+				this.applier.CurrentCoordinatesMutex.Unlock()
+				return chk, err
+			}
+			this.applier.CurrentCoordinatesMutex.Unlock()
+			time.Sleep(500 * time.Millisecond)
+		}
+	}
+}
+
+func (this *Migrator) checkpointLoop() {
+	if this.migrationContext.Noop {
+		this.migrationContext.Log.Debugf("Noop operation; not really checkpointing")
+		return
+	}
+	ticker := time.NewTicker(checkpointInterval)
+	for t := range ticker.C {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
+			return
+		}
+		this.migrationContext.Log.Infof("starting checkpoint at %+v", t)
+		ctx, cancel := context.WithTimeout(context.Background(), checkpointTimeout)
+		chk, err := this.Checkpoint(ctx)
+		if err != nil {
+			if errors.Is(err, context.DeadlineExceeded) {
+				this.migrationContext.Log.Errorf("checkpoint attempt timed out after %+v", checkpointTimeout)
+			} else {
+				this.migrationContext.Log.Errorf("error attempting checkpoint: %+v", err)
+			}
+		} else {
+			this.migrationContext.Log.Infof("checkpoint success at coords=%+v range_min=%+v range_max=%+v",
+				chk.LastTrxCoords.DisplayString(), chk.IterationRangeMin.String(), chk.IterationRangeMax.String())
+		}
+		cancel()
+	}
+}
+
 // executeWriteFuncs writes data via applier: both the rowcopy and the events backlog.
 // This is where the ghost table gets the data. The function fills the data single-threaded.
 // Both event backlog and rowcopy events are polled; the backlog events have precedence.

go/logic/streamer.go

@@ -24,7 +24,7 @@ type BinlogEventListener struct {
 	async        bool
 	databaseName string
 	tableName    string
-	onDmlEvent   func(event *binlog.BinlogDMLEvent) error
+	onDmlEvent   func(event *binlog.BinlogEntry) error
 }
 
 const (
@@ -60,7 +60,7 @@ func NewEventsStreamer(migrationContext *base.MigrationContext) *EventsStreamer
 
 // AddListener registers a new listener for binlog events, on a per-table basis
 func (this *EventsStreamer) AddListener(
-	async bool, databaseName string, tableName string, onDmlEvent func(event *binlog.BinlogDMLEvent) error) (err error) {
+	async bool, databaseName string, tableName string, onDmlEvent func(event *binlog.BinlogEntry) error) (err error) {
 	this.listenersMutex.Lock()
 	defer this.listenersMutex.Unlock()
 
@@ -82,24 +82,24 @@ func (this *EventsStreamer) AddListener(
 
 // notifyListeners will notify relevant listeners with given DML event. Only
 // listeners registered for changes on the table on which the DML operates are notified.
-func (this *EventsStreamer) notifyListeners(binlogEvent *binlog.BinlogDMLEvent) {
+func (this *EventsStreamer) notifyListeners(binlogEntry *binlog.BinlogEntry) {
 	this.listenersMutex.Lock()
 	defer this.listenersMutex.Unlock()
 
 	for _, listener := range this.listeners {
 		listener := listener
-		if !strings.EqualFold(listener.databaseName, binlogEvent.DatabaseName) {
+		if !strings.EqualFold(listener.databaseName, binlogEntry.DmlEvent.DatabaseName) {
 			continue
 		}
-		if !strings.EqualFold(listener.tableName, binlogEvent.TableName) {
+		if !strings.EqualFold(listener.tableName, binlogEntry.DmlEvent.TableName) {
 			continue
 		}
 		if listener.async {
 			go func() {
-				listener.onDmlEvent(binlogEvent)
+				listener.onDmlEvent(binlogEntry)
 			}()
 		} else {
-			listener.onDmlEvent(binlogEvent)
+			listener.onDmlEvent(binlogEntry)
 		}
 	}
 }
@@ -176,7 +176,7 @@ func (this *EventsStreamer) StreamEvents(canStopStreaming func() bool) error {
 	go func() {
 		for binlogEntry := range this.eventsChannel {
 			if binlogEntry.DmlEvent != nil {
-				this.notifyListeners(binlogEntry.DmlEvent)
+				this.notifyListeners(binlogEntry)
 			}
 		}
 	}()

go/logic/streamer_test.go

@@ -90,8 +90,8 @@ func (suite *EventsStreamerTestSuite) TestStreamEvents() {
 	streamCtx, cancel := context.WithCancel(context.Background())
 
 	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
-	err = streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogDMLEvent) error {
-		dmlEvents = append(dmlEvents, event)
+	err = streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogEntry) error {
+		dmlEvents = append(dmlEvents, event.DmlEvent)
 
 		// Stop once we've collected three events
 		if len(dmlEvents) == 3 {
@@ -165,8 +165,8 @@ func (suite *EventsStreamerTestSuite) TestStreamEventsAutomaticallyReconnects()
 	streamCtx, cancel := context.WithCancel(context.Background())
 
 	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
-	err = streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogDMLEvent) error {
-		dmlEvents = append(dmlEvents, event)
+	err = streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogEntry) error {
+		dmlEvents = append(dmlEvents, event.DmlEvent)
 
 		// Stop once we've collected three events
 		if len(dmlEvents) == 3 {