keeping track of delta rows

on concurrent count(*) this means we re-apply delta onto new estimate

Author: Shlomi Noach

go/base/context.go

@@ -95,6 +95,7 @@ type MigrationContext struct {
 
 	TableEngine               string
 	RowsEstimate              int64
+	RowsDeltaEstimate         int64
 	UsedRowsEstimateMethod    RowsEstimateMethod
 	HasSuperPrivilege         bool
 	OriginalBinlogFormat      string

go/logic/applier.go

@@ -890,7 +890,7 @@ func (this *Applier) ApplyDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) error {
 	// no error
 	atomic.AddInt64(&this.migrationContext.TotalDMLEventsApplied, 1)
 	if this.migrationContext.CountTableRows {
-		atomic.AddInt64(&this.migrationContext.RowsEstimate, rowDelta)
+		atomic.AddInt64(&this.migrationContext.RowsDeltaEstimate, rowDelta)
 	}
 	return nil
 }

go/logic/migrator.go

@@ -962,7 +962,7 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 	elapsedTime := this.migrationContext.ElapsedTime()
 	elapsedSeconds := int64(elapsedTime.Seconds())
 	totalRowsCopied := this.migrationContext.GetTotalRowsCopied()
-	rowsEstimate := atomic.LoadInt64(&this.migrationContext.RowsEstimate)
+	rowsEstimate := atomic.LoadInt64(&this.migrationContext.RowsEstimate) + atomic.LoadInt64(&this.migrationContext.RowsDeltaEstimate)
 	var progressPct float64
 	if rowsEstimate == 0 {
 		progressPct = 100.0