feat(ignore-binlog-events): ignore-over-iteration-range-max-binlog

Author: xiehaopeng

doc/command-line-flags.md

@@ -125,6 +125,12 @@ Why is this behavior configurable? Different workloads have different characteri
 
 Noteworthy is that setting `--dml-batch-size` to higher value _does not_ mean `gh-ost` blocks or waits on writes. The batch size is an upper limit on transaction size, not a minimal one. If `gh-ost` doesn't have "enough" events in the pipe, it does not wait on the binary log, it just writes what it already has. This conveniently suggests that if write load is light enough for `gh-ost` to only see a few events in the binary log at a given time, then it is also light enough for `gh-ost` to apply a fraction of the batch size.
 
+
+### ignore-over-iteration-range-max-binlog
+
+Defaults to false. When binlog unique key value is over `MigrationIterationRangeMaxValues`, and less than `MigrationRangeMaxValues`, the binlog will be ignored. Because the data will be synced by copy chunk. When binlog unique key value is over `MigrationRangeMaxValues` or less than `MigrationIterationRangeMaxValues`, the binlog will be applied. Currently when enabled, this only takes effect for single-column unique index of int type.
+
+
 ### exact-rowcount
 
 A `gh-ost` execution need to copy whatever rows you have in your existing table onto the ghost table. This can and often will be, a large number. Exactly what that number is?

go/base/context.go

@@ -201,7 +201,9 @@ type MigrationContext struct {
 	controlReplicasLagResult               mysql.ReplicationLagResult
 	TotalRowsCopied                        int64
 	TotalDMLEventsApplied                  int64
+	TotalDMLEventsIgnored                  int64
 	DMLBatchSize                           int64
+	IgnoreOverIterationRangeMaxBinlog      bool
 	isThrottled                            bool
 	throttleReason                         string
 	throttleReasonHint                     ThrottleReasonHint

go/cmd/gh-ost/main.go

@@ -109,6 +109,7 @@ func main() {
 	defaultRetries := flag.Int64("default-retries", 60, "Default number of retries for various operations before panicking")
 	cutOverLockTimeoutSeconds := flag.Int64("cut-over-lock-timeout-seconds", 3, "Max number of seconds to hold locks on tables while attempting to cut-over (retry attempted when lock exceeds timeout) or attempting instant DDL")
 	niceRatio := flag.Float64("nice-ratio", 0, "force being 'nice', imply sleep time per chunk time; range: [0.0..100.0]. Example values: 0 is aggressive. 1: for every 1ms spent copying rows, sleep additional 1ms (effectively doubling runtime); 0.7: for every 10ms spend in a rowcopy chunk, spend 7ms sleeping immediately after")
+	flag.BoolVar(&migrationContext.IgnoreOverIterationRangeMaxBinlog, "ignore-over-iteration-range-max-binlog", false, "When binlog unique key value is over MigrationIterationRangeMaxValues, and less than MigrationRangeMaxValues, the binlog will be ignored. Because the data will be synced by copy chunk")
 
 	maxLagMillis := flag.Int64("max-lag-millis", 1500, "replication lag at which to throttle operation")
 	replicationLagQuery := flag.String("replication-lag-query", "", "Deprecated. gh-ost uses an internal, subsecond resolution query")

go/logic/applier.go

@@ -30,18 +30,20 @@ const (
 )
 
 type dmlBuildResult struct {
-	query     string
-	args      []interface{}
-	rowsDelta int64
-	err       error
+	query         string
+	args          []interface{}
+	uniqueKeyArgs []interface{}
+	rowsDelta     int64
+	err           error
 }
 
-func newDmlBuildResult(query string, args []interface{}, rowsDelta int64, err error) *dmlBuildResult {
+func newDmlBuildResult(query string, args []interface{}, uniqueKeyArgs []interface{}, rowsDelta int64, err error) *dmlBuildResult {
 	return &dmlBuildResult{
-		query:     query,
-		args:      args,
-		rowsDelta: rowsDelta,
-		err:       err,
+		query:         query,
+		args:          args,
+		uniqueKeyArgs: uniqueKeyArgs,
+		rowsDelta:     rowsDelta,
+		err:           err,
 	}
 }
 
@@ -129,6 +131,7 @@ func (this *Applier) prepareQueries() (err error) {
 		this.migrationContext.OriginalTableColumns,
 		this.migrationContext.SharedColumns,
 		this.migrationContext.MappedSharedColumns,
+		&this.migrationContext.UniqueKey.Columns,
 	); err != nil {
 		return err
 	}
@@ -1190,12 +1193,12 @@ func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) []*dmlB
 	case binlog.DeleteDML:
 		{
 			query, uniqueKeyArgs, err := this.dmlDeleteQueryBuilder.BuildQuery(dmlEvent.WhereColumnValues.AbstractValues())
-			return []*dmlBuildResult{newDmlBuildResult(query, uniqueKeyArgs, -1, err)}
+			return []*dmlBuildResult{newDmlBuildResult(query, uniqueKeyArgs, uniqueKeyArgs, -1, err)}
 		}
 	case binlog.InsertDML:
 		{
-			query, sharedArgs, err := this.dmlInsertQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues())
-			return []*dmlBuildResult{newDmlBuildResult(query, sharedArgs, 1, err)}
+			query, sharedArgs, uniqueKeyArgs, err := this.dmlInsertQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues())
+			return []*dmlBuildResult{newDmlBuildResult(query, sharedArgs, uniqueKeyArgs, 1, err)}
 		}
 	case binlog.UpdateDML:
 		{
@@ -1211,12 +1214,87 @@ func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) []*dmlB
 			args := sqlutils.Args()
 			args = append(args, sharedArgs...)
 			args = append(args, uniqueKeyArgs...)
-			return []*dmlBuildResult{newDmlBuildResult(query, args, 0, err)}
+			return []*dmlBuildResult{newDmlBuildResult(query, args, uniqueKeyArgs, 0, err)}
 		}
 	}
 	return []*dmlBuildResult{newDmlBuildResultError(fmt.Errorf("Unknown dml event type: %+v", dmlEvent.DML))}
 }
 
+// IsIgnoreOverMaxChunkRangeEvent returns true if this event can be ignored, because the data will be synced by copy chunk
+// min       rangeMax        max
+// the value > rangeMax and value < max, ignore = true
+// otherwise ignore = false
+func (this *Applier) IsIgnoreOverMaxChunkRangeEvent(uniqueKeyArgs []interface{}) (bool, error) {
+	if !this.migrationContext.IgnoreOverIterationRangeMaxBinlog {
+		return false, nil
+	}
+
+	// Currently only supports single-column unique index of int type
+	uniqueKeyCols := this.migrationContext.UniqueKey.Columns.Columns()
+	if len(uniqueKeyCols) != 1 {
+		return false, nil
+	}
+	uniqueKeyCol := uniqueKeyCols[0]
+	if uniqueKeyCol.CompareValueFunc == nil {
+		return false, nil
+	}
+
+	// Compare whether it is less than the MigrationIterationRangeMaxValues boundary value. If it is, it cannot be ignored and the corresponding binlog needs to be applied.
+	ignore, err := func() (bool, error) {
+		compareValues := this.migrationContext.MigrationIterationRangeMaxValues
+		if compareValues == nil {
+			// It means that the migration has not started yet, use MigrationRangeMinValues instead
+			compareValues = this.migrationContext.MigrationRangeMinValues
+		}
+
+		than, err := uniqueKeyCol.CompareValueFunc(uniqueKeyArgs[0], compareValues.StringColumn(0))
+		if err != nil {
+			return false, err
+		}
+
+		switch {
+		case than > 0:
+			return true, nil
+		case than < 0:
+			return false, nil
+		default:
+			// Since rowcopy is left-open-right-closed, when it is equal to the MigrationIterationRangeMaxValues boundary value, it cannot be ignored.
+			return false, nil
+		}
+	}()
+	if err != nil {
+		return false, err
+	}
+
+	if !ignore {
+		return false, nil
+	}
+
+	// Compare whether it is greater than the MigrationRangeMaxValues boundary value. If it is, it cannot be ignored and the corresponding binlog needs to be applied.
+	ignore, err = func() (bool, error) {
+		compareValues := this.migrationContext.MigrationRangeMaxValues
+		than, err := uniqueKeyCol.CompareValueFunc(uniqueKeyArgs[0], compareValues)
+		if err != nil {
+			return false, err
+		}
+
+		switch {
+		case than < 0:
+			return true, nil
+		case than > 0:
+			return false, nil
+		default:
+			// Since rowcopy is left-open-right-closed, when it is equal to the MigrationRangeMaxValues boundary value, it can be ignored.
+			return true, nil
+		}
+	}()
+	if err != nil {
+		return false, err
+	}
+
+	return ignore, nil
+}
+
 // ApplyDMLEventQueries applies multiple DML queries onto the _ghost_ table
 func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent)) error {
 	var totalDelta int64
@@ -1244,17 +1322,33 @@ func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent))
 			return err
 		}
 
+		var ignoredEventSize int64
 		buildResults := make([]*dmlBuildResult, 0, len(dmlEvents))
 		nArgs := 0
 		for _, dmlEvent := range dmlEvents {
 			for _, buildResult := range this.buildDMLEventQuery(dmlEvent) {
 				if buildResult.err != nil {
 					return rollback(buildResult.err)
 				}
+				if ignore, err := this.IsIgnoreOverMaxChunkRangeEvent(buildResult.uniqueKeyArgs); err != nil {
+					return rollback(err)
+				} else if ignore {
+					ignoredEventSize++
+					continue
+				}
 				nArgs += len(buildResult.args)
 				buildResults = append(buildResults, buildResult)
 			}
 		}
+		atomic.AddInt64(&this.migrationContext.TotalDMLEventsIgnored, ignoredEventSize)
+
+		// If there are no statements to execute, return directly
+		if len(buildResults) == 0 {
+			if err := tx.Commit(); err != nil {
+				return err
+			}
+			return nil
+		}
 
 		// We batch together the DML queries into multi-statements to minimize network trips.
 		// We have to use the raw driver connection to access the rows affected

go/logic/applier_test.go

@@ -8,6 +8,8 @@ package logic
 import (
 	"context"
 	gosql "database/sql"
+	"fmt"
+	"math/big"
 	"strings"
 	"testing"
 
@@ -181,6 +183,63 @@ func TestApplierBuildDMLEventQuery(t *testing.T) {
 	})
 }
 
+func TestIsIgnoreOverMaxChunkRangeEvent(t *testing.T) {
+	migrationContext := base.NewMigrationContext()
+	migrationContext.IgnoreOverIterationRangeMaxBinlog = true
+	uniqueColumns := sql.NewColumnList([]string{"id"})
+	uniqueColumns.SetColumnCompareValueFunc("id", func(a interface{}, b interface{}) (int, error) {
+		_a := new(big.Int)
+		if _a, _ = _a.SetString(fmt.Sprintf("%+v", a), 10); a == nil {
+			return 0, fmt.Errorf("CompareValueFunc err, %+v convert int is nil", a)
+		}
+		_b := new(big.Int)
+		if _b, _ = _b.SetString(fmt.Sprintf("%+v", b), 10); b == nil {
+			return 0, fmt.Errorf("CompareValueFunc err, %+v convert int is nil", b)
+		}
+		return _a.Cmp(_b), nil
+	})
+
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:    "PRIMARY KEY",
+		Columns: *uniqueColumns,
+	}
+	migrationContext.MigrationRangeMinValues = sql.ToColumnValues([]interface{}{10})
+	migrationContext.MigrationRangeMaxValues = sql.ToColumnValues([]interface{}{123456})
+	migrationContext.MigrationIterationRangeMaxValues = sql.ToColumnValues([]interface{}{11111})
+
+	applier := NewApplier(migrationContext)
+
+	t.Run("less than MigrationRangeMinValues", func(t *testing.T) {
+		ignore, err := applier.IsIgnoreOverMaxChunkRangeEvent([]interface{}{5})
+		require.NoError(t, err)
+		require.False(t, ignore)
+	})
+
+	t.Run("equal to MigrationIterationRangeMaxValues", func(t *testing.T) {
+		ignore, err := applier.IsIgnoreOverMaxChunkRangeEvent([]interface{}{11111})
+		require.NoError(t, err)
+		require.False(t, ignore)
+	})
+
+	t.Run("ignore event", func(t *testing.T) {
+		ignore, err := applier.IsIgnoreOverMaxChunkRangeEvent([]interface{}{88888})
+		require.NoError(t, err)
+		require.True(t, ignore)
+	})
+
+	t.Run("equal to MigrationRangeMaxValues", func(t *testing.T) {
+		ignore, err := applier.IsIgnoreOverMaxChunkRangeEvent([]interface{}{123456})
+		require.NoError(t, err)
+		require.True(t, ignore)
+	})
+
+	t.Run("larger than MigrationRangeMaxValues", func(t *testing.T) {
+		ignore, err := applier.IsIgnoreOverMaxChunkRangeEvent([]interface{}{123457})
+		require.NoError(t, err)
+		require.False(t, ignore)
+	})
+}
+
 func TestApplierInstantDDL(t *testing.T) {
 	migrationContext := base.NewMigrationContext()
 	migrationContext.DatabaseName = "test"

go/logic/inspect.go

@@ -10,6 +10,7 @@ import (
 	gosql "database/sql"
 	"errors"
 	"fmt"
+	"math/big"
 	"reflect"
 	"strings"
 	"sync/atomic"
@@ -635,6 +636,19 @@ func (this *Inspector) applyColumnTypes(databaseName, tableName string, columnsL
 				continue
 			}
 
+			if strings.Contains(columnType, "int") {
+				column.CompareValueFunc = func(a interface{}, b interface{}) (int, error) {
+					_a := new(big.Int)
+					if _a, _ = _a.SetString(fmt.Sprintf("%+v", a), 10); a == nil {
+						return 0, fmt.Errorf("CompareValueFunc err, %+v convert int is nil", a)
+					}
+					_b := new(big.Int)
+					if _b, _ = _b.SetString(fmt.Sprintf("%+v", b), 10); b == nil {
+						return 0, fmt.Errorf("CompareValueFunc err, %+v convert int is nil", b)
+					}
+					return _a.Cmp(_b), nil
+				}
+			}
 			if strings.Contains(columnType, "unsigned") {
 				column.IsUnsigned = true
 			}

go/logic/migrator.go

@@ -1054,9 +1054,10 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 
 	currentBinlogCoordinates := *this.eventsStreamer.GetCurrentBinlogCoordinates()
 
-	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, HeartbeatLag: %.2fs, State: %s; ETA: %s",
+	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Ignored: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, HeartbeatLag: %.2fs, State: %s; ETA: %s",
 		totalRowsCopied, rowsEstimate, progressPct,
 		atomic.LoadInt64(&this.migrationContext.TotalDMLEventsApplied),
+		atomic.LoadInt64(&this.migrationContext.TotalDMLEventsIgnored),
 		len(this.applyEventsQueue), cap(this.applyEventsQueue),
 		base.PrettifyDurationOutput(elapsedTime), base.PrettifyDurationOutput(this.migrationContext.ElapsedRowCopyTime()),
 		currentBinlogCoordinates,

go/sql/builder.go

@@ -466,15 +466,15 @@ func (b *DMLDeleteQueryBuilder) BuildQuery(args []interface{}) (string, []interf
 // DMLInsertQueryBuilder can build INSERT queries for DML events.
 // It holds the prepared query statement so it doesn't need to be recreated every time.
 type DMLInsertQueryBuilder struct {
-	tableColumns, sharedColumns *ColumnList
-	preparedStatement           string
+	tableColumns, sharedColumns, uniqueKeyColumns *ColumnList
+	preparedStatement                             string
 }
 
 // NewDMLInsertQueryBuilder creates a new DMLInsertQueryBuilder.
 // It prepares the INSERT query statement.
 // Returns an error if no shared columns are given, the shared columns are not a subset of the table columns,
 // or the prepared statement cannot be built.
-func NewDMLInsertQueryBuilder(databaseName, tableName string, tableColumns, sharedColumns, mappedSharedColumns *ColumnList) (*DMLInsertQueryBuilder, error) {
+func NewDMLInsertQueryBuilder(databaseName, tableName string, tableColumns, sharedColumns, mappedSharedColumns *ColumnList, uniqueKeyColumns *ColumnList) (*DMLInsertQueryBuilder, error) {
 	if !sharedColumns.IsSubsetOf(tableColumns) {
 		return nil, fmt.Errorf("shared columns is not a subset of table columns in NewDMLInsertQueryBuilder")
 	}
@@ -505,24 +505,33 @@ func NewDMLInsertQueryBuilder(databaseName, tableName string, tableColumns, shar
 	return &DMLInsertQueryBuilder{
 		tableColumns:      tableColumns,
 		sharedColumns:     sharedColumns,
+		uniqueKeyColumns:  uniqueKeyColumns,
 		preparedStatement: stmt,
 	}, nil
 }
 
 // BuildQuery builds the arguments array for a DML event INSERT query.
 // It returns the query string and the shared arguments array.
 // Returns an error if the number of arguments differs from the number of table columns.
-func (b *DMLInsertQueryBuilder) BuildQuery(args []interface{}) (string, []interface{}, error) {
+func (b *DMLInsertQueryBuilder) BuildQuery(args []interface{}) (string, []interface{}, []interface{}, error) {
 	if len(args) != b.tableColumns.Len() {
-		return "", nil, fmt.Errorf("args count differs from table column count in BuildDMLInsertQuery")
+		return "", nil, nil, fmt.Errorf("args count differs from table column count in BuildDMLInsertQuery")
 	}
+
 	sharedArgs := make([]interface{}, 0, b.sharedColumns.Len())
 	for _, column := range b.sharedColumns.Columns() {
 		tableOrdinal := b.tableColumns.Ordinals[column.Name]
 		arg := column.convertArg(args[tableOrdinal], false)
 		sharedArgs = append(sharedArgs, arg)
 	}
-	return b.preparedStatement, sharedArgs, nil
+
+	uniqueKeyArgs := make([]interface{}, 0, b.uniqueKeyColumns.Len())
+	for _, column := range b.uniqueKeyColumns.Columns() {
+		tableOrdinal := b.tableColumns.Ordinals[column.Name]
+		arg := column.convertArg(args[tableOrdinal], true)
+		uniqueKeyArgs = append(uniqueKeyArgs, arg)
+	}
+	return b.preparedStatement, sharedArgs, uniqueKeyArgs, nil
 }
 
 // DMLUpdateQueryBuilder can build UPDATE queries for DML events.