Multithreaded replication WIP

go/base/context.go

@@ -161,6 +161,9 @@ type MigrationContext struct {
 	CutOverType                  CutOver
 	ReplicaServerId              uint
 
+	// Number of workers used by the Coordinator
+	NumWorkers int
+
 	Hostname                               string
 	AssumeMasterHostname                   string
 	ApplierTimeZone                        string

go/binlog/gomysql_reader.go

@@ -6,6 +6,7 @@
 package binlog
 
 import (
+	"bytes"
 	"fmt"
 	"sync"
 
@@ -76,7 +77,7 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate
 	return &returnCoordinates
 }
 
-// StreamEvents
+// handleRowsEvents processes a RowEvent from the binlog and sends the DML event to the entriesChannel.
 func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
 	if this.currentCoordinates.IsLogPosOverflowBeyond4Bytes(&this.LastAppliedRowsEventHint) {
 		return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
@@ -105,6 +106,7 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 		)
 		switch dml {
 		case InsertDML:
+
 			{
 				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
 			}
@@ -128,6 +130,190 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 	return nil
 }
 
+// RowsEventToBinlogEntry processes MySQL RowsEvent into our BinlogEntry for later application.
+// copied from handleRowEvents
+func RowsEventToBinlogEntry(eventType replication.EventType, rowsEvent *replication.RowsEvent, binlogCoords mysql.BinlogCoordinates) (*BinlogEntry, error) {
+	dml := ToEventDML(eventType.String())
+	if dml == NotDML {
+		return nil, fmt.Errorf("Unknown DML type: %s", eventType.String())
+	}
+	binlogEntry := NewBinlogEntryAt(binlogCoords)
+	for i, row := range rowsEvent.Rows {
+		if dml == UpdateDML && i%2 == 1 {
+			// An update has two rows (WHERE+SET)
+			// We do both at the same time
+			continue
+		}
+		binlogEntry.DmlEvent = NewBinlogDMLEvent(
+			string(rowsEvent.Table.Schema),
+			string(rowsEvent.Table.Table),
+			dml,
+		)
+		switch dml {
+		case InsertDML:
+			{
+				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
+			}
+		case UpdateDML:
+			{
+				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(rowsEvent.Rows[i+1])
+			}
+		case DeleteDML:
+			{
+				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+			}
+		}
+	}
+	return binlogEntry, nil
+}
+
+type Transaction struct {
+	SequenceNumber int64
+	LastCommitted  int64
+	Changes        chan *BinlogEntry
+}
+
+func (this *GoMySQLReader) StreamTransactions(ctx context.Context, transactionsChannel chan<- *Transaction) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	previousSequenceNumber := int64(0)
+
+groups:
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		ev, err := this.binlogStreamer.GetEvent(ctx)
+		if err != nil {
+			return err
+		}
+		func() {
+			this.currentCoordinatesMutex.Lock()
+			defer this.currentCoordinatesMutex.Unlock()
+			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
+			this.currentCoordinates.EventSize = int64(ev.Header.EventSize)
+		}()
+
+		fmt.Printf("Event: %s\n", ev.Header.EventType)
+
+		// Read each event and do something with it
+		//
+		// First, ignore all events until we find the next GTID event so that we can start
+		// at a transaction boundary.
+		//
+		// Once we find a GTID event, we can start an event group,
+		// and then process all events in that group.
+		// An event group is defined as all events that are part of the same transaction,
+		// which is defined as all events between the GTID event, a `QueryEvent` containing a `BEGIN` query and ends with
+		// either a XIDEvent or a `QueryEvent` containing a `COMMIT` or `ROLLBACK` query.
+		//
+		// Each group is a struct containing the SequenceNumber, LastCommitted, and a channel of events.
+		//
+		// Once the group has ended, we can start looking for the next GTID event.
+
+		var group *Transaction
+		switch binlogEvent := ev.Event.(type) {
+		case *replication.GTIDEvent:
+			this.migrationContext.Log.Infof("GTIDEvent: %+v", binlogEvent)
+
+			// Bail out if we find a gap in the sequence numbers
+			if previousSequenceNumber != 0 && binlogEvent.SequenceNumber != previousSequenceNumber+1 {
+				return fmt.Errorf("unexpected sequence number: %d, expected %d", binlogEvent.SequenceNumber, previousSequenceNumber+1)
+			}
+
+			group = &Transaction{
+				SequenceNumber: binlogEvent.SequenceNumber,
+				LastCommitted:  binlogEvent.LastCommitted,
+				Changes:        make(chan *BinlogEntry, 1000),
+			}
+
+			previousSequenceNumber = binlogEvent.SequenceNumber
+
+			// We are good to send the transaction, the transaction events arrive async
+			this.migrationContext.Log.Infof("sending transaction: %d %d", group.SequenceNumber, group.LastCommitted)
+			transactionsChannel <- group
+		default:
+			this.migrationContext.Log.Infof("Ignoring Event: %+v", ev.Event)
+			continue
+		}
+
+		// Next event should be a query event
+
+		ev, err = this.binlogStreamer.GetEvent(ctx)
+		if err != nil {
+			close(group.Changes)
+			return err
+		}
+		this.migrationContext.Log.Infof("1 - Event: %s", ev.Header.EventType)
+
+		switch binlogEvent := ev.Event.(type) {
+		case *replication.QueryEvent:
+			if bytes.Equal([]byte("BEGIN"), binlogEvent.Query) {
+				this.migrationContext.Log.Infof("BEGIN for transaction in schema %s", binlogEvent.Schema)
+			} else {
+				this.migrationContext.Log.Infof("QueryEvent: %+v", binlogEvent)
+				this.migrationContext.Log.Infof("Query: %s", binlogEvent.Query)
+
+				close(group.Changes)
+
+				// wait for the next event group
+				continue groups
+			}
+		default:
+			this.migrationContext.Log.Infof("unexpected Event: %+v", ev.Event)
+			close(group.Changes)
+
+			// TODO: handle the group - we want to make sure we process the group's LastCommitted and SequenceNumber
+
+			// wait for the next event group
+			continue groups
+		}
+
+		// Next event should be a table map event
+
+	events:
+		// Now we can start processing the group
+		for {
+			ev, err = this.binlogStreamer.GetEvent(ctx)
+			if err != nil {
+				close(group.Changes)
+				return err
+			}
+			this.migrationContext.Log.Infof("3 - Event: %s", ev.Header.EventType)
+
+			switch binlogEvent := ev.Event.(type) {
+			case *replication.TableMapEvent:
+				this.migrationContext.Log.Infof("TableMapEvent for %s.%s: %+v", binlogEvent.Schema, binlogEvent.Table, binlogEvent)
+			case *replication.RowsEvent:
+				binlogEntry, err := RowsEventToBinlogEntry(ev.Header.EventType, binlogEvent, this.currentCoordinates)
+				if err != nil {
+					close(group.Changes)
+					return err
+				}
+				this.migrationContext.Log.Infof("RowsEvent: %v", binlogEvent)
+				group.Changes <- binlogEntry
+				this.migrationContext.Log.Infof("Length of group.Changes: %d", len(group.Changes))
+			case *replication.XIDEvent:
+				this.migrationContext.Log.Infof("XIDEvent: %+v", binlogEvent)
+				this.migrationContext.Log.Infof("COMMIT for transaction")
+				break events
+			default:
+				close(group.Changes)
+				this.migrationContext.Log.Infof("unexpected Event: %+v", ev.Event)
+				return fmt.Errorf("unexpected Event: %+v", ev.Event)
+			}
+		}
+
+		close(group.Changes)
+
+		this.migrationContext.Log.Infof("done processing group - %d events", len(group.Changes))
+	}
+}
+
 // StreamEvents
 func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
 	if canStopStreaming() {

go/logic/applier.go

@@ -75,11 +75,13 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
 	}
 }
 
-func (this *Applier) InitDBConnections() (err error) {
+func (this *Applier) InitDBConnections(maxConns int) (err error) {
 	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, applierUri); err != nil {
 		return err
 	}
+	this.db.SetMaxOpenConns(maxConns)
+	this.db.SetMaxIdleConns(maxConns)
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if this.singletonDB, _, err = mysql.GetDB(this.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err
@@ -1231,6 +1233,7 @@ func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent))
 					return rollback(buildResult.err)
 				}
 				result, err := tx.Exec(buildResult.query, buildResult.args...)
+
 				if err != nil {
 					err = fmt.Errorf("%w; query=%s; args=%+v", err, buildResult.query, buildResult.args)
 					return rollback(err)

go/logic/applier_test.go

@@ -284,7 +284,7 @@
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	suite.Require().Equal("8.0.40", migrationContext.ApplierMySQLVersion)
@@ -318,7 +318,7 @@
 	suite.Require().NoError(applier.prepareQueries())
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	rc, _, err := suite.mysqlContainer.Exec(ctx, []string{"mysql", "-uroot", "-proot-password", "-e", "CREATE TABLE test._testing_gho (id INT, item_id INT);"})
@@ -380,7 +380,7 @@
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -426,7 +426,7 @@
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
 	err = applier.InitDBConnections()
 	suite.Require().NoError(err)
 
 	chunkSize, rowsAffected, duration, err := applier.ApplyIterationInsertQuery()
@@ -492,7 +492,7 @@
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
 	err = applier.InitDBConnections()
 	suite.Require().NoError(err)
 
 	// Lock one of the rows