Merge branch 'master' of https://github.com/github/gh-ost into arthur/tweak-testcontainers

Author: arthurschreiber

go/logic/applier.go

@@ -60,6 +60,10 @@ type Applier struct {
 	migrationContext  *base.MigrationContext
 	finishedMigrating int64
 	name              string
+
+	dmlDeleteQueryBuilder *sql.DMLDeleteQueryBuilder
+	dmlInsertQueryBuilder *sql.DMLInsertQueryBuilder
+	dmlUpdateQueryBuilder *sql.DMLUpdateQueryBuilder
 }
 
 func NewApplier(migrationContext *base.MigrationContext) *Applier {
@@ -106,6 +110,37 @@ func (this *Applier) InitDBConnections() (err error) {
 	return nil
 }
 
+func (this *Applier) prepareQueries() (err error) {
+	if this.dmlDeleteQueryBuilder, err = sql.NewDMLDeleteQueryBuilder(
+		this.migrationContext.DatabaseName,
+		this.migrationContext.GetGhostTableName(),
+		this.migrationContext.OriginalTableColumns,
+		&this.migrationContext.UniqueKey.Columns,
+	); err != nil {
+		return err
+	}
+	if this.dmlInsertQueryBuilder, err = sql.NewDMLInsertQueryBuilder(
+		this.migrationContext.DatabaseName,
+		this.migrationContext.GetGhostTableName(),
+		this.migrationContext.OriginalTableColumns,
+		this.migrationContext.SharedColumns,
+		this.migrationContext.MappedSharedColumns,
+	); err != nil {
+		return err
+	}
+	if this.dmlUpdateQueryBuilder, err = sql.NewDMLUpdateQueryBuilder(
+		this.migrationContext.DatabaseName,
+		this.migrationContext.GetGhostTableName(),
+		this.migrationContext.OriginalTableColumns,
+		this.migrationContext.SharedColumns,
+		this.migrationContext.MappedSharedColumns,
+		&this.migrationContext.UniqueKey.Columns,
+	); err != nil {
+		return err
+	}
+	return nil
+}
+
 // validateAndReadGlobalVariables potentially reads server global variables, such as the time_zone and wait_timeout.
 func (this *Applier) validateAndReadGlobalVariables() error {
 	query := `select /* gh-ost */ @@global.time_zone, @@global.wait_timeout`
@@ -631,6 +666,8 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		this.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
 		this.migrationContext.GetIteration() == 0,
 		this.migrationContext.IsTransactionalTable(),
+		// TODO: Don't hardcode this
+		strings.HasPrefix(this.migrationContext.ApplierMySQLVersion, "8."),
 	)
 	if err != nil {
 		return chunkSize, rowsAffected, duration, err
@@ -1135,35 +1172,36 @@ func (this *Applier) updateModifiesUniqueKeyColumns(dmlEvent *binlog.BinlogDMLEv
 
 // buildDMLEventQuery creates a query to operate on the ghost table, based on an intercepted binlog
 // event entry on the original table.
-func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) (results [](*dmlBuildResult)) {
+func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) []*dmlBuildResult {
 	switch dmlEvent.DML {
 	case binlog.DeleteDML:
 		{
-			query, uniqueKeyArgs, err := sql.BuildDMLDeleteQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, &this.migrationContext.UniqueKey.Columns, dmlEvent.WhereColumnValues.AbstractValues())
-			return append(results, newDmlBuildResult(query, uniqueKeyArgs, -1, err))
+			query, uniqueKeyArgs, err := this.dmlDeleteQueryBuilder.BuildQuery(dmlEvent.WhereColumnValues.AbstractValues())
+			return []*dmlBuildResult{newDmlBuildResult(query, uniqueKeyArgs, -1, err)}
 		}
 	case binlog.InsertDML:
 		{
-			query, sharedArgs, err := sql.BuildDMLInsertQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, dmlEvent.NewColumnValues.AbstractValues())
-			return append(results, newDmlBuildResult(query, sharedArgs, 1, err))
+			query, sharedArgs, err := this.dmlInsertQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues())
+			return []*dmlBuildResult{newDmlBuildResult(query, sharedArgs, 1, err)}
 		}
 	case binlog.UpdateDML:
 		{
 			if _, isModified := this.updateModifiesUniqueKeyColumns(dmlEvent); isModified {
+				results := make([]*dmlBuildResult, 0, 2)
 				dmlEvent.DML = binlog.DeleteDML
 				results = append(results, this.buildDMLEventQuery(dmlEvent)...)
 				dmlEvent.DML = binlog.InsertDML
 				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 := this.dmlUpdateQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
 			args := sqlutils.Args()
 			args = append(args, sharedArgs...)
 			args = append(args, uniqueKeyArgs...)
-			return append(results, newDmlBuildResult(query, args, 0, err))
+			return []*dmlBuildResult{newDmlBuildResult(query, args, 0, err)}
 		}
 	}
-	return append(results, newDmlBuildResultError(fmt.Errorf("Unknown dml event type: %+v", dmlEvent.DML)))
+	return []*dmlBuildResult{newDmlBuildResultError(fmt.Errorf("Unknown dml event type: %+v", dmlEvent.DML))}
 }
 
 // ApplyDMLEventQueries applies multiple DML queries onto the _ghost_ table

go/logic/applier_test.go

@@ -99,6 +99,7 @@ func TestApplierBuildDMLEventQuery(t *testing.T) {
 	columnValues := sql.ToColumnValues([]interface{}{123456, 42})
 
 	migrationContext := base.NewMigrationContext()
+	migrationContext.DatabaseName = "test"
 	migrationContext.OriginalTableName = "test"
 	migrationContext.OriginalTableColumns = columns
 	migrationContext.SharedColumns = columns
@@ -109,6 +110,7 @@ func TestApplierBuildDMLEventQuery(t *testing.T) {
 	}
 
 	applier := NewApplier(migrationContext)
+	applier.prepareQueries()
 
 	t.Run("delete", func(t *testing.T) {
 		binlogEvent := &binlog.BinlogDMLEvent{
@@ -302,8 +304,13 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
 	migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "item_id"})
 	migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "item_id"})
 	migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "item_id"})
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:    "primary_key",
+		Columns: *sql.NewColumnList([]string{"id"}),
+	}
 
 	applier := NewApplier(migrationContext)
+	suite.Require().NoError(applier.prepareQueries())
 	defer applier.Teardown()
 
 	err = applier.InitDBConnections()

go/logic/migrator.go

@@ -386,6 +386,10 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.inspector.inspectOriginalAndGhostTables(); err != nil {
 		return err
 	}
+	// We can prepare some of the queries on the applier
+	if err := this.applier.prepareQueries(); err != nil {
+		return err
+	}
 	// Validation complete! We're good to execute this migration
 	if err := this.hooksExecutor.onValidated(); err != nil {
 		return err