Merge pull request #41 from github/master-pos-wait-fix2

Various problem fixes

build.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 #
 #
-RELEASE_VERSION="0.7.4"
+RELEASE_VERSION="0.7.16"
 
 buildpath=/tmp/gh-ost
 target=gh-ost

go/base/context.go

@@ -39,9 +39,10 @@ type MigrationContext struct {
 	OriginalTableName string
 	AlterStatement    string
 
-	CountTableRows          bool
-	AllowedRunningOnMaster  bool
-	SwitchToRowBinlogFormat bool
+	CountTableRows           bool
+	AllowedRunningOnMaster   bool
+	SwitchToRowBinlogFormat  bool
+	NullableUniqueKeyAllowed bool
 
 	config      ContextConfig
 	configMutex *sync.Mutex

go/binlog/binlog_reader.go

@@ -5,8 +5,14 @@
 
 package binlog
 
+import (
+	"github.com/github/gh-ost/go/mysql"
+)
+
 // BinlogReader is a general interface whose implementations can choose their methods of reading
 // a binary log file and parsing it into binlog entries
 type BinlogReader interface {
 	StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error
+	GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates
+	Reconnect() error
 }

go/binlog/gomysql_reader.go

@@ -23,11 +23,12 @@ const (
 )
 
 type GoMySQLReader struct {
-	connectionConfig   *mysql.ConnectionConfig
-	binlogSyncer       *replication.BinlogSyncer
-	binlogStreamer     *replication.BinlogStreamer
-	tableMap           map[uint64]string
-	currentCoordinates mysql.BinlogCoordinates
+	connectionConfig       *mysql.ConnectionConfig
+	binlogSyncer           *replication.BinlogSyncer
+	binlogStreamer         *replication.BinlogStreamer
+	tableMap               map[uint64]string
+	currentCoordinates     mysql.BinlogCoordinates
+	lastHandledCoordinates mysql.BinlogCoordinates
 }
 
 func NewGoMySQLReader(connectionConfig *mysql.ConnectionConfig) (binlogReader *GoMySQLReader, err error) {
@@ -39,24 +40,91 @@ func NewGoMySQLReader(connectionConfig *mysql.ConnectionConfig) (binlogReader *G
 	}
 	binlogReader.binlogSyncer = replication.NewBinlogSyncer(serverId, "mysql")
 
-	// Register slave, the MySQL master is at 127.0.0.1:3306, with user root and an empty password
-	err = binlogReader.binlogSyncer.RegisterSlave(connectionConfig.Key.Hostname, uint16(connectionConfig.Key.Port), connectionConfig.User, connectionConfig.Password)
-	if err != nil {
-		return binlogReader, err
-	}
-
 	return binlogReader, err
 }
 
 // ConnectBinlogStreamer
 func (this *GoMySQLReader) ConnectBinlogStreamer(coordinates mysql.BinlogCoordinates) (err error) {
+	if coordinates.IsEmpty() {
+		return log.Errorf("Emptry coordinates at ConnectBinlogStreamer()")
+	}
+	log.Infof("Registering replica at %+v:%+v", this.connectionConfig.Key.Hostname, uint16(this.connectionConfig.Key.Port))
+	if err := this.binlogSyncer.RegisterSlave(this.connectionConfig.Key.Hostname, uint16(this.connectionConfig.Key.Port), this.connectionConfig.User, this.connectionConfig.Password); err != nil {
+		return err
+	}
+
 	this.currentCoordinates = coordinates
+	log.Infof("Connecting binlog streamer at %+v", this.currentCoordinates)
 	// Start sync with sepcified binlog file and position
-	this.binlogStreamer, err = this.binlogSyncer.StartSync(gomysql.Position{coordinates.LogFile, uint32(coordinates.LogPos)})
+	this.binlogStreamer, err = this.binlogSyncer.StartSync(gomysql.Position{this.currentCoordinates.LogFile, uint32(this.currentCoordinates.LogPos)})
 
 	return err
 }
 
+func (this *GoMySQLReader) Reconnect() error {
+	this.binlogSyncer.Close()
+
+	connectCoordinates := &this.lastHandledCoordinates
+	if connectCoordinates.IsEmpty() {
+		connectCoordinates = &this.currentCoordinates
+	}
+	if err := this.ConnectBinlogStreamer(*connectCoordinates); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates {
+	return &this.currentCoordinates
+}
+
+// StreamEvents
+func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
+	if this.currentCoordinates.SmallerThanOrEquals(&this.lastHandledCoordinates) && !this.lastHandledCoordinates.IsEmpty() {
+		log.Infof("Skipping handled query at %+v", this.currentCoordinates)
+		return nil
+	}
+
+	dml := ToEventDML(ev.Header.EventType.String())
+	if dml == NotDML {
+		return fmt.Errorf("Unknown DML type: %s", ev.Header.EventType.String())
+	}
+	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 := NewBinlogEntryAt(this.currentCoordinates)
+		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)
+			}
+		}
+		// The channel will do the throttling. Whoever is reding from the channel
+		// decides whether action is taken sycnhronously (meaning we wait before
+		// next iteration) or asynchronously (we keep pushing more events)
+		// In reality, reads will be synchronous
+		entriesChannel <- binlogEntry
+	}
+	return nil
+}
+
 // StreamEvents
 func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
 	for {
@@ -77,44 +145,11 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 			// future I should remove this.
 			this.tableMap[tableMapEvent.TableID] = string(tableMapEvent.Table)
 		} else if rowsEvent, ok := ev.Event.(*replication.RowsEvent); ok {
-			dml := ToEventDML(ev.Header.EventType.String())
-			if dml == NotDML {
-				return fmt.Errorf("Unknown DML type: %s", ev.Header.EventType.String())
-			}
-			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 := NewBinlogEntryAt(this.currentCoordinates)
-				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)
-					}
-				}
-				// The channel will do the throttling. Whoever is reding from the channel
-				// decides whether action is taken sycnhronously (meaning we wait before
-				// next iteration) or asynchronously (we keep pushing more events)
-				// In reality, reads will be synchronous
-				entriesChannel <- binlogEntry
+			if err := this.handleRowsEvent(ev, rowsEvent, entriesChannel); err != nil {
+				return err
 			}
 		}
+		this.lastHandledCoordinates = this.currentCoordinates
 	}
 	log.Debugf("done streaming events")
 

go/cmd/gh-ost/main.go

@@ -32,6 +32,7 @@ func main() {
 	flag.StringVar(&migrationContext.AlterStatement, "alter", "", "alter statement (mandatory)")
 	flag.BoolVar(&migrationContext.CountTableRows, "exact-rowcount", false, "actually count table rows as opposed to estimate them (results in more accurate progress estimation)")
 	flag.BoolVar(&migrationContext.AllowedRunningOnMaster, "allow-on-master", false, "allow this migration to run directly on master. Preferably it would run on a replica")
+	flag.BoolVar(&migrationContext.NullableUniqueKeyAllowed, "allow-nullable-unique-key", false, "allow gh-ost to migrate based on a unique key with nullable columns. As long as no NULL values exist, this should be OK. If NULL values exist in chosen key, data may be corrupted. Use at your own risk!")
 
 	executeFlag := flag.Bool("execute", false, "actually execute the alter & migrate the table. Default is noop: do some tests and exit")
 	flag.BoolVar(&migrationContext.TestOnReplica, "test-on-replica", false, "Have the migration run on a replica, not on the master. At the end of migration tables are not swapped; gh-ost issues `STOP SLAVE` and you can compare the two tables for building trust")