Count rows queries

build.sh

@@ -2,8 +2,8 @@
 #
 #
 
-RELEASE_VERSION=
-buildpath=
+RELEASE_VERSION=2
+buildpath=/tmp
 
 function setuptree() {
   b=$( mktemp -d $buildpath/gh-ostXXXXXX ) || return 1

go/base/context.go

@@ -123,6 +123,7 @@ type MigrationContext struct {
 	MaxLagMillisecondsThrottleThreshold int64
 	throttleControlReplicaKeys          *mysql.InstanceKeyMap
 	ThrottleFlagFile                    string
+	copyWhereClause                     string
 	ThrottleAdditionalFlagFile          string
 	throttleQuery                       string
 	throttleHTTP                        string
@@ -725,6 +726,19 @@ func (this *MigrationContext) GetCriticalLoad() LoadMap {
 	return this.criticalLoad.Duplicate()
 }
 
+func (this *MigrationContext) GetWhereClause() string {
+	this.throttleMutex.Lock()
+	defer this.throttleMutex.Unlock()
+
+	return this.copyWhereClause
+}
+
+func (this *MigrationContext) SetWhereClause(WhereClause string) {
+	this.throttleMutex.Lock()
+	defer this.throttleMutex.Unlock()
+	this.copyWhereClause = WhereClause
+}
+
 func (this *MigrationContext) GetNiceRatio() float64 {
 	this.throttleMutex.Lock()
 	defer this.throttleMutex.Unlock()

go/cmd/gh-ost/main.go

@@ -115,6 +115,7 @@ func main() {
 	throttleControlReplicas := flag.String("throttle-control-replicas", "", "List of replicas on which to check for lag; comma delimited. Example: myhost1.com:3306,myhost2.com,myhost3.com:3307")
 	throttleQuery := flag.String("throttle-query", "", "when given, issued (every second) to check if operation should throttle. Expecting to return zero for no-throttle, >0 for throttle. Query is issued on the migrated server. Make sure this query is lightweight")
 	throttleHTTP := flag.String("throttle-http", "", "when given, gh-ost checks given URL via HEAD request; any response code other than 200 (OK) causes throttling; make sure it has low latency response")
+	copyWhereClause := flag.String("where-clause", "1=1", "added where clause for the insert query, filtering table rows")
 	flag.Int64Var(&migrationContext.ThrottleHTTPIntervalMillis, "throttle-http-interval-millis", 100, "Number of milliseconds to wait before triggering another HTTP throttle check")
 	flag.Int64Var(&migrationContext.ThrottleHTTPTimeoutMillis, "throttle-http-timeout-millis", 1000, "Number of milliseconds to use as an HTTP throttle check timeout")
 	ignoreHTTPErrors := flag.Bool("ignore-http-errors", false, "ignore HTTP connection errors during throttle check")
@@ -303,6 +304,7 @@ func main() {
 	migrationContext.SetThrottleHTTP(*throttleHTTP)
 	migrationContext.SetIgnoreHTTPErrors(*ignoreHTTPErrors)
 	migrationContext.SetDefaultNumRetries(*defaultRetries)
+	migrationContext.SetWhereClause(*copyWhereClause)
 	migrationContext.ApplyCredentials()
 	if err := migrationContext.SetupTLS(); err != nil {
 		migrationContext.Log.Fatale(err)

go/logic/applier.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"time"
+	"context"
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/binlog"
@@ -631,13 +632,20 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		this.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
 		this.migrationContext.GetIteration() == 0,
 		this.migrationContext.IsTransactionalTable(),
+		this.migrationContext.GetWhereClause(),
 	)
 	if err != nil {
 		return chunkSize, rowsAffected, duration, err
 	}
 
 	sqlResult, err := func() (gosql.Result, error) {
-		tx, err := this.db.Begin()
+		var conn *gosql.Conn
+		conn, err = this.db.Conn(context.Background())
+		if (conn == nil || err != nil) {
+		    fmt.Sprintf("failed to get connection")
+		    return nil, err
+		}
+		tx, err := conn.BeginTx(context.Background(), nil)
 		if err != nil {
 			return nil, err
 		}
@@ -646,16 +654,17 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s'`, this.migrationContext.ApplierTimeZone)
 		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, this.generateSqlModeQuery())
 
-		if _, err := tx.Exec(sessionQuery); err != nil {
+		if _, err := tx.ExecContext(context.Background(), sessionQuery); err != nil {
 			return nil, err
 		}
-		result, err := tx.Exec(query, explodedArgs...)
+		result, err := tx.ExecContext(context.Background(), query, explodedArgs...)
 		if err != nil {
 			return nil, err
 		}
 		if err := tx.Commit(); err != nil {
 			return nil, err
 		}
+		conn.Close()
 		return result, nil
 	}()
 

go/logic/inspect.go

@@ -554,7 +554,7 @@ func (this *Inspector) CountTableRows(ctx context.Context) error {
 		return err
 	}
 
-	query := fmt.Sprintf(`select /* gh-ost */ count(*) as count_rows from %s.%s`, sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
+	query := fmt.Sprintf(`select /* gh-ost */ table_rows as count_rows from information_schema.tables where table_schema="%s" and table_name="%s"`, sql.EscapeName(this.migrationContext.DatabaseName), sql.EscapeName(this.migrationContext.OriginalTableName))
 	var rowsEstimate int64
 	if err := conn.QueryRowContext(ctx, query).Scan(&rowsEstimate); err != nil {
 		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {

go/logic/migrator.go

@@ -842,13 +842,14 @@ func (this *Migrator) printMigrationStatusHint(writers ...io.Writer) {
 	)
 	maxLoad := this.migrationContext.GetMaxLoad()
 	criticalLoad := this.migrationContext.GetCriticalLoad()
-	fmt.Fprintf(w, "# chunk-size: %+v; max-lag-millis: %+vms; dml-batch-size: %+v; max-load: %s; critical-load: %s; nice-ratio: %f\n",
+	fmt.Fprintf(w, "# chunk-size: %+v; max-lag-millis: %+vms; dml-batch-size: %+v; max-load: %s; critical-load: %s; nice-ratio: %f; where-clause: %s\n",
 		atomic.LoadInt64(&this.migrationContext.ChunkSize),
 		atomic.LoadInt64(&this.migrationContext.MaxLagMillisecondsThrottleThreshold),
 		atomic.LoadInt64(&this.migrationContext.DMLBatchSize),
 		maxLoad.String(),
 		criticalLoad.String(),
 		this.migrationContext.GetNiceRatio(),
+		this.migrationContext.GetWhereClause(),
 	)
 	if this.migrationContext.ThrottleFlagFile != "" {
 		setIndicator := ""

go/sql/builder.go

@@ -182,7 +182,7 @@ func BuildRangePreparedComparison(columns *ColumnList, args []interface{}, compa
 	return BuildRangeComparison(columns.Names(), values, args, comparisonSign)
 }
 
-func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartValues, rangeEndValues []string, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool) (result string, explodedArgs []interface{}, err error) {
+func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartValues, rangeEndValues []string, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool, whereClause string) (result string, explodedArgs []interface{}, err error) {
 	if len(sharedColumns) == 0 {
 		return "", explodedArgs, fmt.Errorf("Got 0 shared columns in BuildRangeInsertQuery")
 	}
@@ -232,19 +232,19 @@ func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName strin
 				%s.%s
 			force index (%s)
 			where
-				(%s and %s)
+				(%s and %s and %s)
 				%s
 		)`,
 		databaseName, originalTableName, databaseName, ghostTableName, mappedSharedColumnsListing,
 		sharedColumnsListing, databaseName, originalTableName, uniqueKey,
-		rangeStartComparison, rangeEndComparison, transactionalClause)
+		rangeStartComparison, rangeEndComparison, whereClause, transactionalClause)
 	return result, explodedArgs, nil
 }
 
-func BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool) (result string, explodedArgs []interface{}, err error) {
+func BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool, whereClause string) (result string, explodedArgs []interface{}, err error) {
 	rangeStartValues := buildColumnsPreparedValues(uniqueKeyColumns)
 	rangeEndValues := buildColumnsPreparedValues(uniqueKeyColumns)
-	return BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, includeRangeStartValues, transactionalTable)
+	return BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, includeRangeStartValues, transactionalTable, whereClause)
 }
 
 func BuildUniqueKeyRangeEndPreparedQueryViaOffset(databaseName, tableName string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, chunkSize int64, includeRangeStartValues bool, hint string) (result string, explodedArgs []interface{}, err error) {