Removed StateTracker complexity

batch_writer.go

@@ -10,7 +10,7 @@ import (
 type BatchWriter struct {
 	DB             *sql.DB
 	InlineVerifier *InlineVerifier
-	StateTracker   *CopyStateTracker
+	StateTracker   *StateTracker
 
 	DatabaseRewrites map[string]string
 	TableRewrites    map[string]string

binlog_writer.go

@@ -17,7 +17,7 @@ type BinlogWriter struct {
 	WriteRetries int
 
 	ErrorHandler ErrorHandler
-	StateTracker *CopyStateTracker
+	StateTracker *StateTracker
 
 	binlogEventBuffer chan DMLEvent
 	logger            *logrus.Entry
@@ -110,7 +110,7 @@ func (b *BinlogWriter) writeEvents(events []DMLEvent) error {
 	}
 
 	if b.StateTracker != nil {
-		b.StateTracker.UpdateLastProcessedBinlogPosition(events[len(events)-1].BinlogPosition())
+		b.StateTracker.UpdateLastWrittenBinlogPosition(events[len(events)-1].BinlogPosition())
 	}
 
 	return nil

data_iterator.go

@@ -16,7 +16,7 @@ type DataIterator struct {
 
 	ErrorHandler ErrorHandler
 	CursorConfig *CursorConfig
-	StateTracker *CopyStateTracker
+	StateTracker *StateTracker
 
 	targetPKs      *sync.Map
 	batchListeners []func(*RowBatch) error
@@ -32,7 +32,7 @@ func (d *DataIterator) Run(tables []*TableSchema) {
 	// tracking state. However, some methods are still useful so we initialize
 	// a minimal local instance.
 	if d.StateTracker == nil {
-		d.StateTracker = NewCopyStateTracker(0)
+		d.StateTracker = NewStateTracker(0)
 	}
 
 	d.logger.WithField("tablesCount", len(tables)).Info("starting data iterator run")

ferry.go

@@ -102,7 +102,7 @@ func (f *Ferry) NewDataIterator() *DataIterator {
 			BatchSize:   f.Config.DataIterationBatchSize,
 			ReadRetries: f.Config.DBReadRetries,
 		},
-		StateTracker: f.StateTracker.CopyStage,
+		StateTracker: f.StateTracker,
 	}
 
 	if f.CopyFilter != nil {
@@ -144,7 +144,7 @@ func (f *Ferry) NewBinlogWriter() *BinlogWriter {
 		WriteRetries: f.Config.DBWriteRetries,
 
 		ErrorHandler: f.ErrorHandler,
-		StateTracker: f.StateTracker.CopyStage,
+		StateTracker: f.StateTracker,
 	}
 }
 
@@ -159,7 +159,7 @@ func (f *Ferry) NewBatchWriter() *BatchWriter {
 
 	batchWriter := &BatchWriter{
 		DB:           f.TargetDB,
-		StateTracker: f.StateTracker.CopyStage,
+		StateTracker: f.StateTracker,
 
 		DatabaseRewrites: f.Config.DatabaseRewrites,
 		TableRewrites:    f.Config.TableRewrites,
@@ -480,7 +480,8 @@ func (f *Ferry) Start() error {
 	// If we don't set this now, there is a race condition where Ghostferry
 	// is terminated with some rows copied but no binlog events are written.
 	// This guarentees that we are able to restart from a valid location.
-	f.StateTracker.CopyStage.UpdateLastProcessedBinlogPosition(pos)
+	f.StateTracker.UpdateLastWrittenBinlogPosition(pos)
+	f.StateTracker.UpdateLastStoredBinlogPositionForInlineVerifier(pos)
 
 	return nil
 }
@@ -698,7 +699,7 @@ func (f *Ferry) Progress() *Progress {
 	s.FinalBinlogPos = f.BinlogStreamer.targetBinlogPosition
 
 	// Table Progress
-	serializedState := f.StateTracker.CopyStage.Serialize()
+	serializedState := f.StateTracker.Serialize(nil)
 	s.Tables = make(map[string]TableProgress)
 	targetPKs := make(map[string]uint64)
 	f.DataIterator.targetPKs.Range(func(k, v interface{}) bool {
@@ -731,7 +732,7 @@ func (f *Ferry) Progress() *Progress {
 	// ETA
 	var totalPKsToCopy uint64 = 0
 	var completedPKs uint64 = 0
-	estimatedPKsPerSecond := f.StateTracker.CopyStage.EstimatedPKsPerSecond()
+	estimatedPKsPerSecond := f.StateTracker.EstimatedPKsPerSecond()
 	for _, targetPK := range targetPKs {
 		totalPKsToCopy += targetPK
 	}

sharding/test/callbacks_test.go

@@ -134,7 +134,7 @@ func (t *CallbacksTestSuite) TestFailsRunOnPanicError() {
 		stateDump := &ghostferry.SerializableState{}
 		jsonErr = json.Unmarshal([]byte(errorData["StateDump"]), stateDump)
 		t.Require().Nil(jsonErr)
-		t.Require().NotNil(stateDump.CopyStage)
+		t.Require().NotNil(stateDump)
 		t.Require().True(len(stateDump.LastKnownTableSchemaCache) > 0)
 
 		w.WriteHeader(http.StatusInternalServerError)