Add estimated row batch size in bytes to state

batch_writer.go

@@ -123,7 +123,7 @@ func (w *BatchWriter) WriteRowBatch(batch *RowBatch) error {
 		// Note that the state tracker expects us the track based on the original
 		// database and table names as opposed to the target ones.
 		if w.StateTracker != nil {
-			w.StateTracker.UpdateLastSuccessfulPaginationKey(batch.TableSchema().String(), endPaginationKeypos, uint64(batch.Size()))
+			w.StateTracker.UpdateLastSuccessfulPaginationKey(batch.TableSchema().String(), endPaginationKeypos, uint64(batch.Size()), batch.EstimateByteSize())
 		}
 
 		return nil

cursor.go

@@ -69,7 +69,6 @@ type Cursor struct {
 
 	paginationKeyColumn         *schema.TableColumn
 	lastSuccessfulPaginationKey uint64
-	rowsExamined                uint64
 	logger                      *logrus.Entry
 }
 
@@ -142,7 +141,6 @@ func (c *Cursor) Each(f func(*RowBatch) error) error {
 		tx.Rollback()
 
 		c.lastSuccessfulPaginationKey = paginationKeypos
-		c.rowsExamined += uint64(batch.Size())
 	}
 
 	return nil

ferry.go

@@ -846,7 +846,7 @@ func (f *Ferry) Progress() *Progress {
 	serializedState := f.StateTracker.Serialize(nil, nil)
 	// Note the below will not necessarily be synchronized with serializedState.
 	// This is fine as we don't need to be super precise with performance data.
-	rowsWrittenPerTable := f.StateTracker.RowsWrittenPerTable()
+	rowStatsWrittenPerTable := f.StateTracker.RowStatsWrittenPerTable()
 
 	s.Tables = make(map[string]TableProgress)
 	targetPaginationKeys := make(map[string]uint64)
@@ -876,13 +876,14 @@ func (f *Ferry) Progress() *Progress {
 			currentAction = TableActionWaiting
 		}
 
-		rowsWritten, _ := rowsWrittenPerTable[tableName]
+		rowWrittenStats, _ := rowStatsWrittenPerTable[tableName]
 
 		s.Tables[tableName] = TableProgress{
 			LastSuccessfulPaginationKey: lastSuccessfulPaginationKey,
 			TargetPaginationKey:         targetPaginationKeys[tableName],
 			CurrentAction:               currentAction,
-			RowsWritten:                 rowsWritten,
+			RowsWritten:                 rowWrittenStats.NumRows,
+			BytesWritten: 					     rowWrittenStats.NumBytes,
 		}
 	}
 

progress.go

@@ -15,6 +15,7 @@ type TableProgress struct {
 	TargetPaginationKey         uint64
 	CurrentAction               string // Possible values are defined via the constants TableAction*
 	RowsWritten                 uint64
+	BytesWritten 								uint64
 }
 
 type Progress struct {

row_batch.go

@@ -2,6 +2,7 @@ package ghostferry
 
 import (
 	"strings"
+	"encoding/json"
 )
 
 type RowBatch struct {
@@ -23,6 +24,19 @@ func (e *RowBatch) Values() []RowData {
 	return e.values
 }
 
+func (e *RowBatch) EstimateByteSize() uint64 {
+	var total int
+	for _, v := range e.values {
+		size, err := json.Marshal(v)
+		if err != nil {
+			continue
+		}
+		total += len(size)
+	}
+
+	return uint64(total)
+}
+
 func (e *RowBatch) PaginationKeyIndex() int {
 	return e.paginationKeyIndex
 }

state_tracker.go

@@ -79,6 +79,10 @@ func newSpeedLogRing(speedLogCount int) *ring.Ring {
 	return speedLog
 }
 
+type RowStats struct {
+	NumRows  uint64
+	NumBytes uint64
+}
 type StateTracker struct {
 	BinlogRWMutex *sync.RWMutex
 	CopyRWMutex   *sync.RWMutex
@@ -92,8 +96,8 @@ type StateTracker struct {
 
 	// TODO: Performance tracking should be refactored out of the state tracker,
 	// as it confuses the focus of this struct.
-	iterationSpeedLog   *ring.Ring
-	rowsWrittenPerTable map[string]uint64
+	iterationSpeedLog       *ring.Ring
+	rowStatsWrittenPerTable map[string]RowStats
 }
 
 func NewStateTracker(speedLogCount int) *StateTracker {
@@ -104,7 +108,7 @@ func NewStateTracker(speedLogCount int) *StateTracker {
 		lastSuccessfulPaginationKeys: make(map[string]uint64),
 		completedTables:              make(map[string]bool),
 		iterationSpeedLog:            newSpeedLogRing(speedLogCount),
-		rowsWrittenPerTable:          make(map[string]uint64),
+		rowStatsWrittenPerTable:      make(map[string]RowStats),
 	}
 }
 
@@ -141,7 +145,7 @@ func (s *StateTracker) UpdateLastResumableBinlogPositionForTargetVerifier(pos my
 	s.lastStoredBinlogPositionForTargetVerifier = pos
 }
 
-func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginationKey uint64, rowsWrittenForThisBatch uint64) {
+func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginationKey uint64, rowsWrittenForThisBatch uint64, bytesWrittenForThisBatch uint64) {
 	s.CopyRWMutex.Lock()
 	defer s.CopyRWMutex.Unlock()
 
@@ -152,17 +156,21 @@ func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginatio
 	// hopefully will motivate us to fix it by refactoring the state tracker a bit
 	// in the future. Namely, the tracking of performance metrics and the tracking
 	// of pagination key locations should be done more separately than it is now.
-	s.rowsWrittenPerTable[table] += rowsWrittenForThisBatch
+	prev := s.rowStatsWrittenPerTable[table]
+	rowsTotal := prev.NumRows + rowsWrittenForThisBatch
+	bytesTotal := prev.NumBytes + bytesWrittenForThisBatch
+
+	s.rowStatsWrittenPerTable[table] = RowStats{NumRows: rowsTotal, NumBytes: bytesTotal}
 
 	s.updateSpeedLog(deltaPaginationKey)
 }
 
-func (s *StateTracker) RowsWrittenPerTable() map[string]uint64 {
+func (s *StateTracker) RowStatsWrittenPerTable() map[string]RowStats {
 	s.CopyRWMutex.RLock()
 	defer s.CopyRWMutex.RUnlock()
 
-	d := make(map[string]uint64)
-	for k, v := range s.rowsWrittenPerTable {
+	d := make(map[string]RowStats)
+	for k, v := range s.rowStatsWrittenPerTable {
 		d[k] = v
 	}
 

test/integration/callbacks_test.rb

@@ -25,6 +25,9 @@ def test_progress_callback
     assert count > 0, "There should be some rows on the target, not 0."
     assert_equal count, progress.last["Tables"]["gftest.test_table_1"]["RowsWritten"]
 
+    # data column is 32 characters so each row should be at least 32 bytes
+    assert count * 32 < progress.last["Tables"]["gftest.test_table_1"]["BytesWritten"], "Each row should have more than 32 bytes"
+
     assert_equal 0, progress.last["ActiveDataIterators"]
 
     refute progress.last["LastSuccessfulBinlogPos"]["Name"].nil?