Fixes from feedback

data_iterator.go

@@ -47,7 +47,14 @@ func (d *DataIterator) Run() {
 	}
 
 	for table, maxPk := range tablesWithData {
-		d.targetPKs.Store(table.String(), maxPk)
+		tableName := table.String()
+		if d.StateTracker.IsTableComplete(tableName) {
+			// In a previous run, the table may have been completed.
+			// We don't need to reiterate those tables as it has already been done.
+			delete(tablesWithData, tableName)
+		} else {
+			d.targetPKs.Store(table.String(), maxPk)
+		}
 	}
 
 	tablesQueue := make(chan *schema.Table)

ferry.go

@@ -116,12 +116,6 @@ func (f *Ferry) NewBinlogStreamer() *BinlogStreamer {
 	}
 }
 
-// Even though this function is identical to NewBinlogStreamer, it is here
-// for consistency so it will lead to less confusion.
-func (f *Ferry) NewBinlogStreamerWithoutStateTracker() *BinlogStreamer {
-	return f.NewBinlogStreamer()
-}
-
 func (f *Ferry) NewBinlogWriter() *BinlogWriter {
 	return &BinlogWriter{
 		DB:               f.TargetDB,

state_tracker.go

@@ -69,6 +69,13 @@ func (s *StateTracker) MarkTableAsCompleted(table string) {
 	s.completedTables[table] = true
 }
 
+func (s *StateTracker) IsTableComplete(table string) {
+	s.tableMutex.Lock()
+	defer s.tableMutex.Unlock()
+
+	return s.completedTables[table]
+}
+
 func (s *StateTracker) UpdateLastWrittenBinlogPosition(pos mysql.Position) {
 	s.binlogMutex.Lock()
 	defer s.binlogMutex.Unlock()