relay: update BinlogFile/Pos along with Gtid

when full copy done on applier.

Author: ffffwh

internal/client/driver/mysql/applier.go

@@ -451,6 +451,7 @@ func (a *Applier) executeWriteFuncs() {
 				a.logger.Printf("mysql.applier: Rows copy complete.number of rows:%d", a.mysqlContext.TotalRowsReplay)
 				a.mysqlContext.Gtid = a.currentCoordinates.RetrievedGtidSet
 				a.mysqlContext.BinlogFile = a.currentCoordinates.File
+				a.mysqlContext.BinlogPos = a.currentCoordinates.Position
 				break
 			}
 			if a.shutdown {
@@ -901,6 +902,9 @@ func (a *Applier) initiateStreaming() error {
 			a.onError(TaskStateDead, err)
 		}
 		a.currentCoordinates.RetrievedGtidSet = dumpData.Gtid
+		a.currentCoordinates.File = dumpData.LogFile
+		a.currentCoordinates.Position = dumpData.LogPos
+
 		a.mysqlContext.Stage = models.StageSlaveWaitingForWorkersToProcessQueue
 
 		for atomic.LoadInt64(&a.nDumpEntry) != 0 {

internal/client/driver/mysql/dumper.go

@@ -81,6 +81,8 @@ func NewDumper(db usql.QueryAble, table *config.Table, chunkSize int64,
 type dumpStatResult struct {
 	Gtid       string
 	TotalCount int64
+	LogFile    string
+	LogPos     int64
 }
 
 type DumpEntryOrig struct {

internal/client/driver/mysql/extractor.go

@@ -294,7 +294,12 @@ func (e *Extractor) Run() {
 			e.onError(TaskStateDead, err)
 			return
 		}
-		dumpMsg, err := Encode(&dumpStatResult{Gtid: e.initialBinlogCoordinates.GtidSet, TotalCount: e.mysqlContext.RowsEstimate})
+		dumpMsg, err := Encode(&dumpStatResult{
+			Gtid: e.initialBinlogCoordinates.GtidSet,
+			LogFile: e.initialBinlogCoordinates.LogFile,
+			LogPos: e.initialBinlogCoordinates.LogPos,
+			TotalCount: e.mysqlContext.RowsEstimate,
+		})
 		if err != nil {
 			e.onError(TaskStateDead, err)
 		}