Check RowsAffected when applying DML events to get more accurate stat…

…istics (#844)

* Check RowsAffected when applying DML events to get more accurate statistics

Addresses #600.

When applying a DML event, check the RowsAffected on the `Result`
struct. Since all DML event queries are point queries, this will only
ever be 0 or 1. The applier then takes this value and multiplies by
the `rowsDelta` of the event, resulting in a properly-signed, accurate
row delta to use in the statistics.

If an error occurs here, log it, but do not surface this as an
actual error .. simply assume the DML affected a row and move on. It
will be inaccurate, but this is already the case.

* Fix import

* update wording to warning log message

Co-authored-by: Tim Vaillancourt <[email protected]>

go/logic/applier.go

@@ -1,5 +1,5 @@
 /*
-   Copyright 2016 GitHub Inc.
+   Copyright 2021 GitHub Inc.
 	 See https://github.com/github/gh-ost/blob/master/LICENSE
 */
 
@@ -8,6 +8,7 @@ package logic
 import (
 	gosql "database/sql"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -16,8 +17,8 @@ import (
 	"github.com/github/gh-ost/go/mysql"
 	"github.com/github/gh-ost/go/sql"
 
-	"github.com/outbrain/golib/sqlutils"
-	"sync"
+	"github.com/openark/golib/log"
+	"github.com/openark/golib/sqlutils"
 )
 
 const (
@@ -1070,11 +1071,20 @@ func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent))
 				if buildResult.err != nil {
 					return rollback(buildResult.err)
 				}
-				if _, err := tx.Exec(buildResult.query, buildResult.args...); err != nil {
+				result, err := tx.Exec(buildResult.query, buildResult.args...)
+				if err != nil {
 					err = fmt.Errorf("%s; query=%s; args=%+v", err.Error(), buildResult.query, buildResult.args)
 					return rollback(err)
 				}
-				totalDelta += buildResult.rowsDelta
+
+				rowsAffected, err := result.RowsAffected()
+				if err != nil {
+					log.Warningf("error getting rows affected from DML event query: %s. i'm going to assume that the DML affected a single row, but this may result in inaccurate statistics", err)
+					rowsAffected = 1
+				}
+				// each DML is either a single insert (delta +1), update (delta +0) or delete (delta -1).
+				// multiplying by the rows actually affected (either 0 or 1) will give an accurate row delta for this DML event
+				totalDelta += buildResult.rowsDelta * rowsAffected
 			}
 		}
 		if err := tx.Commit(); err != nil {