store rows copied & dml applied

Author: meiji163

go/logic/applier.go

@@ -435,6 +435,8 @@ func (this *Applier) CreateCheckpointTable() error {
 		"`gh_ost_chk_timestamp` bigint",
 		"`gh_ost_chk_coords` varchar(4096)",
 		"`gh_ost_chk_iteration` bigint",
+		"`gh_ost_rows_copied` bigint",
+		"`gh_ost_dml_applied` bigint",
 	}
 	for _, col := range this.migrationContext.UniqueKey.Columns.Columns() {
 		if col.MySQLType == "" {
@@ -623,7 +625,7 @@ func (this *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) {
 	if err != nil {
 		return insertId, err
 	}
-	args := sqlutils.Args(chk.LastTrxCoords.String(), chk.Iteration)
+	args := sqlutils.Args(chk.LastTrxCoords.String(), chk.Iteration, chk.RowsCopied, chk.DMLApplied)
 	args = append(args, uniqueKeyArgs...)
 	res, err := this.db.Exec(query, args...)
 	if err != nil {
@@ -641,7 +643,7 @@ func (this *Applier) ReadLastCheckpoint() (*Checkpoint, error) {
 
 	var coordStr string
 	var timestamp int64
-	ptrs := []interface{}{&chk.Id, &timestamp, &coordStr, &chk.Iteration}
+	ptrs := []interface{}{&chk.Id, &timestamp, &coordStr, &chk.Iteration, &chk.RowsCopied, &chk.DMLApplied}
 	ptrs = append(ptrs, chk.IterationRangeMin.ValuesPointers...)
 	ptrs = append(ptrs, chk.IterationRangeMax.ValuesPointers...)
 	err := row.Scan(ptrs...)

go/logic/applier_test.go

@@ -700,6 +700,8 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 		IterationRangeMin: applier.migrationContext.MigrationRangeMinValues,
 		IterationRangeMax: applier.migrationContext.MigrationRangeMaxValues,
 		Iteration:         2,
+		RowsCopied:        100000,
+		DMLApplied:        200000,
 	}
 	id, err := applier.WriteCheckpoint(chk)
 	suite.Require().NoError(err)
@@ -712,6 +714,8 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 	suite.Require().Equal(chk.LastTrxCoords.String(), gotChk.LastTrxCoords.String())
 	suite.Require().Equal(chk.IterationRangeMin.String(), gotChk.IterationRangeMin.String())
 	suite.Require().Equal(chk.IterationRangeMax.String(), gotChk.IterationRangeMax.String())
+	suite.Require().Equal(chk.RowsCopied, gotChk.RowsCopied)
+	suite.Require().Equal(chk.DMLApplied, gotChk.DMLApplied)
 }
 
 func TestApplier(t *testing.T) {

go/logic/checkpoint.go

@@ -26,4 +26,6 @@ type Checkpoint struct {
 	// for the chunk copier range.
 	IterationRangeMax *sql.ColumnValues
 	Iteration         int64
+	RowsCopied        int64
+	DMLApplied        int64
 }

go/logic/migrator.go

@@ -428,6 +428,8 @@ func (this *Migrator) Migrate() (err error) {
 		this.migrationContext.MigrationIterationRangeMinValues = lastCheckpoint.IterationRangeMin
 		this.migrationContext.MigrationIterationRangeMaxValues = lastCheckpoint.IterationRangeMax
 		this.migrationContext.Iteration = lastCheckpoint.Iteration
+		this.migrationContext.TotalRowsCopied = lastCheckpoint.RowsCopied
+		this.migrationContext.TotalDMLEventsApplied = lastCheckpoint.DMLApplied
 		this.migrationContext.InitialStreamerCoords = lastCheckpoint.LastTrxCoords
 		if err := this.initiateStreaming(); err != nil {
 			return err
@@ -1397,7 +1399,7 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 
 // Checkpoint attempts to write a checkpoint of the Migrator's current state.
 // It gets the binlog coordinates of the last received trx and waits until the
-// applier reaches that trx. At that point it's safe to resume from these
+// applier reaches that trx. At that point it's safe to resume from these coordinates.
 func (this *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) {
 	coords := this.eventsStreamer.GetCurrentBinlogCoordinates()
 	this.applier.LastIterationRangeMutex.Lock()
@@ -1410,6 +1412,8 @@ func (this *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) {
 		IterationRangeMin: this.applier.LastIterationRangeMinValues.Clone(),
 		IterationRangeMax: this.applier.LastIterationRangeMaxValues.Clone(),
 		LastTrxCoords:     coords,
+		RowsCopied:        atomic.LoadInt64(&this.migrationContext.TotalRowsCopied),
+		DMLApplied:        atomic.LoadInt64(&this.migrationContext.TotalDMLEventsApplied),
 	}
 	this.applier.LastIterationRangeMutex.Unlock()
 

go/sql/builder.go

@@ -123,9 +123,13 @@ func NewCheckpointQueryBuilder(databaseName, tableName string, uniqueKeyColumns
 	stmt := fmt.Sprintf(`
 		insert /* gh-ost */
 		into %s.%s
-			(gh_ost_chk_timestamp, gh_ost_chk_coords, gh_ost_chk_iteration, %s, %s)
+			(gh_ost_chk_timestamp, gh_ost_chk_coords, gh_ost_chk_iteration,
+			 gh_ost_rows_copied, gh_ost_dml_applied,
+  			 %s, %s)
 		values
-			(unix_timestamp(now()), ?, ?, %s, %s)`,
+			(unix_timestamp(now()), ?, ?,
+			 ?, ?,
+			 %s, %s)`,
 		databaseName, tableName,
 		strings.Join(minUniqueColNames, ", "),
 		strings.Join(maxUniqueColNames, ", "),