Fix lexographical comparison of binlog filenames (#1604)

* Parse binlog file numbers numerically instead of lexicographically to correctly order files like binlog.999999 < binlog.1000000.

Would cause the stream to ignore all incoming events and render the gh-ost process stuck:
https://github.com/github/gh-ost/blob/48b34bcbfde730b2548d598dee98e9c1f0d2fcce/go/binlog/gomysql_reader.go#L85-L88

Possibly remediated by 005043d too, which drops the SmallerThanOrEqual check from GoMySqlReader.handleRowsEvent

* Remove unused fn FileBinlogCoordinates.FileSmallerThan

Author: grodowski

go/mysql/binlog_file.go

@@ -78,13 +78,12 @@ func (this *FileBinlogCoordinates) SmallerThan(other BinlogCoordinates) bool {
 	if !ok || other == nil {
 		return false
 	}
-	if this.LogFile < coord.LogFile {
-		return true
-	}
-	if this.LogFile == coord.LogFile && this.LogPos < coord.LogPos {
-		return true
+
+	fileNumberDist := this.FileNumberDistance(coord)
+	if fileNumberDist == 0 {
+		return this.LogPos < coord.LogPos
 	}
-	return false
+	return fileNumberDist > 0
 }
 
 // SmallerThanOrEquals returns true if this coordinate is the same or equal to the other one.
@@ -100,15 +99,6 @@ func (this *FileBinlogCoordinates) SmallerThanOrEquals(other BinlogCoordinates)
 	return this.LogFile == coord.LogFile && this.LogPos == coord.LogPos // No Type comparison
 }
 
-// FileSmallerThan returns true if this coordinate's file is strictly smaller than the other's.
-func (this *FileBinlogCoordinates) FileSmallerThan(other BinlogCoordinates) bool {
-	coord, ok := other.(*FileBinlogCoordinates)
-	if !ok || other == nil {
-		return false
-	}
-	return this.LogFile < coord.LogFile
-}
-
 // FileNumberDistance returns the numeric distance between this coordinate's file number and the other's.
 // Effectively it means "how many rotates/FLUSHes would make these coordinates's file reach the other's"
 func (this *FileBinlogCoordinates) FileNumberDistance(other *FileBinlogCoordinates) int {

go/mysql/binlog_file_test.go

@@ -134,3 +134,19 @@ func TestIsLogPosOverflowBeyond4Bytes(t *testing.T) {
 		require.True(t, curCoordinates.IsLogPosOverflowBeyond4Bytes(preCoordinates))
 	}
 }
+
+func TestBinlogCoordinates_LogFileZeroPaddedTransition(t *testing.T) {
+	c1 := FileBinlogCoordinates{LogFile: "mysql-bin.999999", LogPos: 100}
+	c2 := FileBinlogCoordinates{LogFile: "mysql-bin.1000000", LogPos: 100}
+
+	require.True(t, c1.SmallerThan(&c2))
+}
+
+func TestBinlogCoordinates_SameLogFileDifferentPosition(t *testing.T) {
+	c1 := FileBinlogCoordinates{LogFile: "binlog.000001", LogPos: 100}
+	c2 := FileBinlogCoordinates{LogFile: "binlog.000001", LogPos: 200}
+
+	require.True(t, c1.SmallerThan(&c2))
+	require.False(t, c2.SmallerThan(&c1))
+	require.False(t, c1.SmallerThan(&c1))
+}