Support --checksum-data flag, on-the-fly checksum verification

.github/workflows/ci.yml

@@ -8,13 +8,18 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@master
+    - uses: actions/checkout@v2
 
-    - name: Set up Go 1.12
+    - name: Set up Go 1.14
       uses: actions/setup-go@v1
       with:
-        version: 1.12
-      id: go
+        go-version: 1.14
 
     - name: Build
       run: script/cibuild
+
+    - name: Upload gh-ost binary artifact
+      uses: actions/upload-artifact@v1
+      with:
+        name: gh-ost
+        path: bin/gh-ost

.github/workflows/replica-tests.yml

@@ -8,13 +8,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@master
+    - uses: actions/checkout@v2
 
-    - name: Set up Go 1.12
+    - name: Set up Go 1.14
       uses: actions/setup-go@v1
       with:
-        version: 1.12
-      id: go
+        go-version: 1.14
 
     - name: migration tests
       run:  script/cibuild-gh-ost-replica-tests

Dockerfile.packaging

@@ -1,6 +1,6 @@
 #
 
-FROM golang:1.12.6
+FROM golang:1.14.4
 
 RUN apt-get update
 RUN apt-get install -y ruby ruby-dev rubygems build-essential

Dockerfile.test

@@ -1,4 +1,4 @@
-FROM golang:1.12.1
+FROM golang:1.14.4
 LABEL maintainer="github@github.com"
 
 RUN apt-get update

README.md

@@ -94,7 +94,7 @@ Please see [Coding gh-ost](doc/coding-ghost.md) for a guide to getting started d
 
 [Download latest release here](https://github.com/github/gh-ost/releases/latest)
 
-`gh-ost` is a Go project; it is built with Go `1.12` and above. To build on your own, use either:
+`gh-ost` is a Go project; it is built with Go `1.14` and above. To build on your own, use either:
 - [script/build](https://github.com/github/gh-ost/blob/master/script/build) - this is the same build script used by CI hence the authoritative; artifact is `./bin/gh-ost` binary.
 - [build.sh](https://github.com/github/gh-ost/blob/master/build.sh) for building `tar.gz` artifacts in `/tmp/gh-ost`
 

build.sh

@@ -18,8 +18,8 @@ function build {
   GOOS=$3
   GOARCH=$4
 
-  if ! go version | egrep -q 'go(1\.1[234])' ; then
-    echo "go version must be 1.12 or above"
+  if ! go version | egrep -q 'go(1\.1[456])' ; then
+    echo "go version must be 1.14 or above"
     exit 1
   fi
 

go/base/checksum_comparison.go

@@ -0,0 +1,49 @@
+/*
+   Copyright 2015 Shlomi Noach, courtesy Booking.com
+	 See https://github.com/github/gh-ost/blob/master/LICENSE
+*/
+
+package base
+
+import (
+	"fmt"
+
+	"github.com/github/gh-ost/go/sql"
+)
+
+type ChecksumFunc func() (checksum string, err error)
+
+// BinlogCoordinates described binary log coordinates in the form of log file & log position.
+type ChecksumComparison struct {
+	Iteration                        int64
+	OriginalTableChecksumFunc        ChecksumFunc
+	GhostTableChecksumFunc           ChecksumFunc
+	MigrationIterationRangeMinValues *sql.ColumnValues
+	MigrationIterationRangeMaxValues *sql.ColumnValues
+	Attempts                         int
+}
+
+func NewChecksumComparison(
+	iteration int64,
+	originalTableChecksumFunc, ghostTableChecksumFunc ChecksumFunc,
+	rangeMinValues, rangeMaxValues *sql.ColumnValues,
+) *ChecksumComparison {
+	return &ChecksumComparison{
+		Iteration:                        iteration,
+		OriginalTableChecksumFunc:        originalTableChecksumFunc,
+		GhostTableChecksumFunc:           ghostTableChecksumFunc,
+		MigrationIterationRangeMinValues: rangeMinValues,
+		MigrationIterationRangeMaxValues: rangeMaxValues,
+		Attempts:                         0,
+	}
+}
+
+func (this *ChecksumComparison) IncrementAttempts() {
+	this.Attempts = this.Attempts + 1
+}
+
+func (this *ChecksumComparison) String() string {
+	return fmt.Sprintf("iteration: %d, range: [%s]..[%s], attempts: %d",
+		this.Iteration, this.MigrationIterationRangeMinValues, this.MigrationIterationRangeMaxValues, this.Attempts,
+	)
+}

go/base/context.go

@@ -77,12 +77,14 @@ func NewThrottleCheckResult(throttle bool, reason string, reasonHint ThrottleRea
 type MigrationContext struct {
 	Uuid string
 
-	DatabaseName      string
-	OriginalTableName string
-	AlterStatement    string
+	DatabaseName          string
+	OriginalTableName     string
+	AlterStatement        string
+	AlterStatementOptions string // anything following the 'ALTER TABLE [schema.]table' from AlterStatement
 
 	CountTableRows           bool
 	ConcurrentCountTableRows bool
+	ChecksumData             bool
 	AllowedRunningOnMaster   bool
 	AllowedMasterMaster      bool
 	SwitchToRowBinlogFormat  bool
@@ -178,6 +180,9 @@ type MigrationContext struct {
 	pointOfInterestTimeMutex               *sync.Mutex
 	CurrentLag                             int64
 	currentProgress                        uint64
+	PendingChecksumComparisons             int64
+	SuccessfulChecksumComparisons          int64
+	SubmittedChecksumComparisons           int64
 	ThrottleHTTPStatusCode                 int64
 	controlReplicasLagResult               mysql.ReplicationLagResult
 	TotalRowsCopied                        int64
@@ -206,6 +211,7 @@ type MigrationContext struct {
 	GhostTableVirtualColumns         *sql.ColumnList
 	GhostTableUniqueKeys             [](*sql.UniqueKey)
 	UniqueKey                        *sql.UniqueKey
+	GhostUniqueKey                   *sql.UniqueKey
 	SharedColumns                    *sql.ColumnList
 	ColumnRenameMap                  map[string]string
 	DroppedColumnsMap                map[string]bool

go/cmd/gh-ost/main.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/logic"
+	"github.com/github/gh-ost/go/sql"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/outbrain/golib/log"
 
@@ -66,6 +67,7 @@ func main() {
 	flag.StringVar(&migrationContext.AlterStatement, "alter", "", "alter statement (mandatory)")
 	flag.BoolVar(&migrationContext.CountTableRows, "exact-rowcount", false, "actually count table rows as opposed to estimate them (results in more accurate progress estimation)")
 	flag.BoolVar(&migrationContext.ConcurrentCountTableRows, "concurrent-rowcount", true, "(with --exact-rowcount), when true (default): count rows after row-copy begins, concurrently, and adjust row estimate later on; when false: first count rows, then start row copy")
+	flag.BoolVar(&migrationContext.ChecksumData, "checksum-data", false, "if true, checksum original and ghost table shared data on the fly, fail migration if checksum mismatches. Checksum queries run on applier node (the master unless testing on replica)")
 	flag.BoolVar(&migrationContext.AllowedRunningOnMaster, "allow-on-master", false, "allow this migration to run directly on master. Preferably it would run on a replica")
 	flag.BoolVar(&migrationContext.AllowedMasterMaster, "allow-master-master", false, "explicitly allow running in a master-master setup")
 	flag.BoolVar(&migrationContext.NullableUniqueKeyAllowed, "allow-nullable-unique-key", false, "allow gh-ost to migrate based on a unique key with nullable columns. As long as no NULL values exist, this should be OK. If NULL values exist in chosen key, data may be corrupted. Use at your own risk!")
@@ -172,14 +174,25 @@ func main() {
 		migrationContext.Log.SetLevel(log.ERROR)
 	}
 
+	if migrationContext.AlterStatement == "" {
+		log.Fatalf("--alter must be provided and statement must not be empty")
+	}
+	parser := sql.NewParserFromAlterStatement(migrationContext.AlterStatement)
+	migrationContext.AlterStatementOptions = parser.GetAlterStatementOptions()
+
 	if migrationContext.DatabaseName == "" {
-		migrationContext.Log.Fatalf("--database must be provided and database name must not be empty")
+		if parser.HasExplicitSchema() {
+			migrationContext.DatabaseName = parser.GetExplicitSchema()
+		} else {
+			log.Fatalf("--database must be provided and database name must not be empty, or --alter must specify database name")
+		}
 	}
 	if migrationContext.OriginalTableName == "" {
-		migrationContext.Log.Fatalf("--table must be provided and table name must not be empty")
-	}
-	if migrationContext.AlterStatement == "" {
-		migrationContext.Log.Fatalf("--alter must be provided and statement must not be empty")
+		if parser.HasExplicitTable() {
+			migrationContext.OriginalTableName = parser.GetExplicitTable()
+		} else {
+			log.Fatalf("--table must be provided and table name must not be empty, or --alter must specify table name")
+		}
 	}
 	migrationContext.Noop = !(*executeFlag)
 	if migrationContext.AllowedRunningOnMaster && migrationContext.TestOnReplica {

go/logic/applier.go

@@ -21,6 +21,7 @@ import (
 
 const (
 	atomicCutOverMagicHint = "ghost-cut-over-sentry"
+	groupConcatMaxLength   = 1024 * 1024
 )
 
 type dmlBuildResult struct {
@@ -68,7 +69,7 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
 
 func (this *Applier) InitDBConnections() (err error) {
 
-	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
+	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName, fmt.Sprintf("group_concat_max_len=%d", groupConcatMaxLength))
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, applierUri); err != nil {
 		return err
 	}
@@ -190,7 +191,7 @@ func (this *Applier) AlterGhost() error {
 	query := fmt.Sprintf(`alter /* gh-ost */ table %s.%s %s`,
 		sql.EscapeName(this.migrationContext.DatabaseName),
 		sql.EscapeName(this.migrationContext.GetGhostTableName()),
-		this.migrationContext.AlterStatement,
+		this.migrationContext.AlterStatementOptions,
 	)
 	this.migrationContext.Log.Infof("Altering ghost table %s.%s",
 		sql.EscapeName(this.migrationContext.DatabaseName),
@@ -454,25 +455,25 @@ func (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange boo
 
 // ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where
 // data actually gets copied from original table.
-func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
+func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected int64, duration time.Duration, checksumComparison *base.ChecksumComparison, err error) {
 	startTime := time.Now()
 	chunkSize = atomic.LoadInt64(&this.migrationContext.ChunkSize)
 
-	query, explodedArgs, err := sql.BuildRangeInsertPreparedQuery(
+	insertQuery, originalChecksumQuery, ghostChecksumQuery, explodedArgs, err := sql.BuildRangeInsertPreparedQuery(
 		this.migrationContext.DatabaseName,
 		this.migrationContext.OriginalTableName,
 		this.migrationContext.GetGhostTableName(),
 		this.migrationContext.SharedColumns.Names(),
 		this.migrationContext.MappedSharedColumns.Names(),
-		this.migrationContext.UniqueKey.Name,
-		&this.migrationContext.UniqueKey.Columns,
+		this.migrationContext.UniqueKey,
+		this.migrationContext.GhostUniqueKey,
 		this.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
 		this.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
 		this.migrationContext.GetIteration() == 0,
 		this.migrationContext.IsTransactionalTable(),
 	)
 	if err != nil {
-		return chunkSize, rowsAffected, duration, err
+		return chunkSize, rowsAffected, duration, checksumComparison, err
 	}
 
 	sqlResult, err := func() (gosql.Result, error) {
@@ -491,7 +492,7 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		if _, err := tx.Exec(sessionQuery); err != nil {
 			return nil, err
 		}
-		result, err := tx.Exec(query, explodedArgs...)
+		result, err := tx.Exec(insertQuery, explodedArgs...)
 		if err != nil {
 			return nil, err
 		}
@@ -502,7 +503,7 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 	}()
 
 	if err != nil {
-		return chunkSize, rowsAffected, duration, err
+		return chunkSize, rowsAffected, duration, checksumComparison, err
 	}
 	rowsAffected, _ = sqlResult.RowsAffected()
 	duration = time.Since(startTime)
@@ -512,7 +513,38 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		this.migrationContext.MigrationIterationRangeMaxValues,
 		this.migrationContext.GetIteration(),
 		chunkSize)
-	return chunkSize, rowsAffected, duration, nil
+
+	var originalTableChecksumFunc base.ChecksumFunc = func() (checksum string, err error) {
+		err = this.db.QueryRow(originalChecksumQuery, explodedArgs...).Scan(&checksum)
+		return checksum, err
+	}
+	var ghostTableChecksumFunc base.ChecksumFunc = func() (checksum string, err error) {
+		err = this.db.QueryRow(ghostChecksumQuery, explodedArgs...).Scan(&checksum)
+		return checksum, err
+	}
+	checksumComparison = base.NewChecksumComparison(
+		this.migrationContext.GetIteration(),
+		originalTableChecksumFunc, ghostTableChecksumFunc,
+		this.migrationContext.MigrationIterationRangeMinValues,
+		this.migrationContext.MigrationIterationRangeMaxValues,
+	)
+
+	return chunkSize, rowsAffected, duration, checksumComparison, nil
+}
+
+func (this *Applier) CompareChecksum(checksumComparison *base.ChecksumComparison) error {
+	originalChecksum, err := checksumComparison.OriginalTableChecksumFunc()
+	if err != nil {
+		return err
+	}
+	ghostChecksum, err := checksumComparison.GhostTableChecksumFunc()
+	if err != nil {
+		return err
+	}
+	if originalChecksum != ghostChecksum {
+		return fmt.Errorf("Checksum failure. Iteration: %d", checksumComparison.Iteration)
+	}
+	return nil
 }
 
 // LockOriginalTable places a write lock on the original table
@@ -811,6 +843,10 @@ func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocke
 	}
 
 	tableLockTimeoutSeconds := this.migrationContext.CutOverLockTimeoutSeconds * 2
+	if this.migrationContext.ChecksumData {
+		// Allow extra time for checksum to evaluate
+		tableLockTimeoutSeconds += this.migrationContext.CutOverLockTimeoutSeconds
+	}
 	this.migrationContext.Log.Infof("Setting LOCK timeout as %d seconds", tableLockTimeoutSeconds)
 	query = fmt.Sprintf(`set session lock_wait_timeout:=%d`, tableLockTimeoutSeconds)
 	if _, err := tx.Exec(query); err != nil {

go/logic/inspect.go

@@ -125,7 +125,7 @@ func (this *Inspector) inspectOriginalAndGhostTables() (err error) {
 	if err != nil {
 		return err
 	}
-	sharedUniqueKeys, err := this.getSharedUniqueKeys(this.migrationContext.OriginalTableUniqueKeys, this.migrationContext.GhostTableUniqueKeys)
+	sharedUniqueKeys, ghostSharedUniqueKeys, err := this.getSharedUniqueKeys(this.migrationContext.OriginalTableUniqueKeys, this.migrationContext.GhostTableUniqueKeys)
 	if err != nil {
 		return err
 	}
@@ -150,13 +150,15 @@ func (this *Inspector) inspectOriginalAndGhostTables() (err error) {
 		}
 		if uniqueKeyIsValid {
 			this.migrationContext.UniqueKey = sharedUniqueKeys[i]
+			this.migrationContext.GhostUniqueKey = ghostSharedUniqueKeys[i]
 			break
 		}
 	}
 	if this.migrationContext.UniqueKey == nil {
 		return fmt.Errorf("No shared unique key can be found after ALTER! Bailing out")
 	}
-	this.migrationContext.Log.Infof("Chosen shared unique key is %s", this.migrationContext.UniqueKey.Name)
+	this.migrationContext.Log.Infof("Chosen shared unique key is %s. ghost unique key is %s", this.migrationContext.UniqueKey.Name, this.migrationContext.GhostUniqueKey.Name)
+	this.migrationContext.Log.Infof("Chosen shared unique key columns %+v. ghost unique key columns: %+v", this.migrationContext.UniqueKey.Columns.Names(), this.migrationContext.GhostUniqueKey.Columns.Names())
 	if this.migrationContext.UniqueKey.HasNullable {
 		if this.migrationContext.NullableUniqueKeyAllowed {
 			this.migrationContext.Log.Warningf("Chosen key (%s) has nullable columns. You have supplied with --allow-nullable-unique-key and so this migration proceeds. As long as there aren't NULL values in this key's column, migration should be fine. NULL values will corrupt migration's data", this.migrationContext.UniqueKey)
@@ -668,17 +670,18 @@ func (this *Inspector) getCandidateUniqueKeys(tableName string) (uniqueKeys [](*
 
 // getSharedUniqueKeys returns the intersection of two given unique keys,
 // testing by list of columns
-func (this *Inspector) getSharedUniqueKeys(originalUniqueKeys, ghostUniqueKeys [](*sql.UniqueKey)) (uniqueKeys [](*sql.UniqueKey), err error) {
+func (this *Inspector) getSharedUniqueKeys(originalUniqueKeys, ghostUniqueKeys [](*sql.UniqueKey)) (sharedUniqueKeys [](*sql.UniqueKey), ghostSharedUniqueKeys [](*sql.UniqueKey), err error) {
 	// We actually do NOT rely on key name, just on the set of columns. This is because maybe
 	// the ALTER is on the name itself...
 	for _, originalUniqueKey := range originalUniqueKeys {
 		for _, ghostUniqueKey := range ghostUniqueKeys {
 			if originalUniqueKey.Columns.EqualsByNames(&ghostUniqueKey.Columns) {
-				uniqueKeys = append(uniqueKeys, originalUniqueKey)
+				sharedUniqueKeys = append(sharedUniqueKeys, originalUniqueKey)
+				ghostSharedUniqueKeys = append(ghostSharedUniqueKeys, ghostUniqueKey)
 			}
 		}
 	}
-	return uniqueKeys, nil
+	return sharedUniqueKeys, ghostSharedUniqueKeys, nil
 }
 
 // getSharedColumns returns the intersection of two lists of columns in same order as the first list