Add --chunk-concurrent-size for parallel row-copy

Port of PR #1398 by @shaohk: allows multiple row-copy chunks to execute
in parallel within each iteration using errgroup.

Key changes:
- Add IterationRangeValues struct for thread-safe range passing
- Serialize range calculation with CalculateNextIterationRangeEndValuesLock
- Rewrite iterateChunks to spawn N goroutines per queue item via errgroup
- Return SQL warnings from ApplyIterationInsertQuery (eliminates race on
  shared MigrationLastInsertSQLWarnings field)
- Increase DB connection pool when concurrency > default pool size
- Add --chunk-concurrent-size CLI flag (default 1, no behavior change)

Co-authored-by: shaohk <shaohk@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: 3 people

doc/command-line-flags.md

@@ -78,6 +78,14 @@ See also: [`resuming-migrations`](resume.md)
 
 `--checkpoint-seconds` specifies the seconds between checkpoints. Default is 300.
 
+### chunk-concurrent-size
+
+`--chunk-concurrent-size=1`, the number of goroutines to execute chunk-copy operations concurrently in each copy time slot. Default `1` (sequential), allowed range `1`-`100`.
+
+When set to a value greater than 1, multiple chunks are calculated and copied in parallel within each write-function invocation. This can significantly speed up row-copy on large tables when MySQL can handle concurrent writes to the ghost table.
+
+Each concurrent chunk calculates its own non-overlapping key range under a serialization lock, so there is no risk of duplicate or overlapping copies.
+
 ### conf
 
 `--conf=/path/to/my.cnf`: file where credentials are specified. Should be in (or contain) the following format:

go/base/context.go

@@ -26,6 +26,16 @@ import (
 	"github.com/go-ini/ini"
 )
 
+// IterationRangeValues holds the range boundaries for a single chunk-copy iteration.
+// Used by concurrent row-copy to pass isolated range values to each worker goroutine.
+type IterationRangeValues struct {
+	Min              *sql.ColumnValues
+	Max              *sql.ColumnValues
+	Size             int64
+	IncludeMinValues bool
+	HasFurtherRange  bool
+}
+
 // RowsEstimateMethod is the type of row number estimation
 type RowsEstimateMethod string
 
@@ -130,6 +140,7 @@ type MigrationContext struct {
 	HeartbeatIntervalMilliseconds       int64
 	defaultNumRetries                   int64
 	ChunkSize                           int64
+	ChunkConcurrentSize                 int64
 	niceRatio                           float64
 	MaxLagMillisecondsThrottleThreshold int64
 	throttleControlReplicaKeys          *mysql.InstanceKeyMap
@@ -237,27 +248,28 @@ type MigrationContext struct {
 	AbortError error
 	abortMutex *sync.Mutex
 
-	OriginalTableColumnsOnApplier    *sql.ColumnList
-	OriginalTableColumns             *sql.ColumnList
-	OriginalTableVirtualColumns      *sql.ColumnList
-	OriginalTableUniqueKeys          [](*sql.UniqueKey)
-	OriginalTableAutoIncrement       uint64
-	GhostTableColumns                *sql.ColumnList
-	GhostTableVirtualColumns         *sql.ColumnList
-	GhostTableUniqueKeys             [](*sql.UniqueKey)
-	UniqueKey                        *sql.UniqueKey
-	SharedColumns                    *sql.ColumnList
-	ColumnRenameMap                  map[string]string
-	DroppedColumnsMap                map[string]bool
-	MappedSharedColumns              *sql.ColumnList
-	MigrationLastInsertSQLWarnings   []string
-	MigrationRangeMinValues          *sql.ColumnValues
-	MigrationRangeMaxValues          *sql.ColumnValues
-	Iteration                        int64
-	MigrationIterationRangeMinValues *sql.ColumnValues
-	MigrationIterationRangeMaxValues *sql.ColumnValues
-	InitialStreamerCoords            mysql.BinlogCoordinates
-	ForceTmpTableName                string
+	OriginalTableColumnsOnApplier            *sql.ColumnList
+	OriginalTableColumns                     *sql.ColumnList
+	OriginalTableVirtualColumns              *sql.ColumnList
+	OriginalTableUniqueKeys                  [](*sql.UniqueKey)
+	OriginalTableAutoIncrement               uint64
+	GhostTableColumns                        *sql.ColumnList
+	GhostTableVirtualColumns                 *sql.ColumnList
+	GhostTableUniqueKeys                     [](*sql.UniqueKey)
+	UniqueKey                                *sql.UniqueKey
+	SharedColumns                            *sql.ColumnList
+	ColumnRenameMap                          map[string]string
+	DroppedColumnsMap                        map[string]bool
+	MappedSharedColumns                      *sql.ColumnList
+	MigrationLastInsertSQLWarnings           []string
+	MigrationRangeMinValues                  *sql.ColumnValues
+	MigrationRangeMaxValues                  *sql.ColumnValues
+	Iteration                                int64
+	MigrationIterationRangeMinValues         *sql.ColumnValues
+	MigrationIterationRangeMaxValues         *sql.ColumnValues
+	CalculateNextIterationRangeEndValuesLock *sync.Mutex
+	InitialStreamerCoords                    mysql.BinlogCoordinates
+	ForceTmpTableName                        string
 
 	IncludeTriggers     bool
 	RemoveTriggerSuffix bool
@@ -307,29 +319,31 @@ type ContextConfig struct {
 func NewMigrationContext() *MigrationContext {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	return &MigrationContext{
-		Uuid:                                uuid.NewString(),
-		defaultNumRetries:                   60,
-		ChunkSize:                           1000,
-		InspectorConnectionConfig:           mysql.NewConnectionConfig(),
-		ApplierConnectionConfig:             mysql.NewConnectionConfig(),
-		MaxLagMillisecondsThrottleThreshold: 1500,
-		CutOverLockTimeoutSeconds:           3,
-		DMLBatchSize:                        10,
-		etaNanoseonds:                       ETAUnknown,
-		maxLoad:                             NewLoadMap(),
-		criticalLoad:                        NewLoadMap(),
-		throttleMutex:                       &sync.Mutex{},
-		throttleHTTPMutex:                   &sync.Mutex{},
-		throttleControlReplicaKeys:          mysql.NewInstanceKeyMap(),
-		configMutex:                         &sync.Mutex{},
-		pointOfInterestTimeMutex:            &sync.Mutex{},
-		lastHeartbeatOnChangelogMutex:       &sync.Mutex{},
-		ColumnRenameMap:                     make(map[string]string),
-		PanicAbort:                          make(chan error),
-		ctx:                                 ctx,
-		cancelFunc:                          cancelFunc,
-		abortMutex:                          &sync.Mutex{},
-		Log:                                 NewDefaultLogger(),
+		Uuid:                                     uuid.NewString(),
+		defaultNumRetries:                        60,
+		ChunkSize:                                1000,
+		ChunkConcurrentSize:                      1,
+		InspectorConnectionConfig:                mysql.NewConnectionConfig(),
+		ApplierConnectionConfig:                  mysql.NewConnectionConfig(),
+		MaxLagMillisecondsThrottleThreshold:      1500,
+		CutOverLockTimeoutSeconds:                3,
+		DMLBatchSize:                             10,
+		etaNanoseonds:                            ETAUnknown,
+		maxLoad:                                  NewLoadMap(),
+		criticalLoad:                             NewLoadMap(),
+		throttleMutex:                            &sync.Mutex{},
+		throttleHTTPMutex:                        &sync.Mutex{},
+		throttleControlReplicaKeys:               mysql.NewInstanceKeyMap(),
+		configMutex:                              &sync.Mutex{},
+		pointOfInterestTimeMutex:                 &sync.Mutex{},
+		lastHeartbeatOnChangelogMutex:            &sync.Mutex{},
+		CalculateNextIterationRangeEndValuesLock: &sync.Mutex{},
+		ColumnRenameMap:                          make(map[string]string),
+		PanicAbort:                               make(chan error),
+		ctx:                                      ctx,
+		cancelFunc:                               cancelFunc,
+		abortMutex:                               &sync.Mutex{},
+		Log:                                      NewDefaultLogger(),
 	}
 }
 
@@ -690,6 +704,13 @@ func (mctx *MigrationContext) SetChunkSize(chunkSize int64) {
 	atomic.StoreInt64(&mctx.ChunkSize, chunkSize)
 }
 
+func (mctx *MigrationContext) SetChunkConcurrentSize(chunkConcurrentSize int64) {
+	if chunkConcurrentSize < 1 {
+		chunkConcurrentSize = 1
+	}
+	atomic.StoreInt64(&mctx.ChunkConcurrentSize, chunkConcurrentSize)
+}
+
 func (mctx *MigrationContext) SetDMLBatchSize(batchSize int64) {
 	if batchSize < 1 {
 		batchSize = 1

go/cmd/gh-ost/main.go

@@ -107,6 +107,7 @@ func main() {
 	flag.BoolVar(&migrationContext.CutOverExponentialBackoff, "cut-over-exponential-backoff", false, "Wait exponentially longer intervals between failed cut-over attempts. Wait intervals obey a maximum configurable with 'exponential-backoff-max-interval').")
 	exponentialBackoffMaxInterval := flag.Int64("exponential-backoff-max-interval", 64, "Maximum number of seconds to wait between attempts when performing various operations with exponential backoff.")
 	chunkSize := flag.Int64("chunk-size", 1000, "amount of rows to handle in each iteration (allowed range: 10-100,000)")
+	chunkConcurrentSize := flag.Int64("chunk-concurrent-size", 1, "number of goroutines to execute chunks concurrently in each copy time slot (range 1-100)")
 	dmlBatchSize := flag.Int64("dml-batch-size", 10, "batch size for DML events to apply in a single transaction (range 1-1000)")
 	defaultRetries := flag.Int64("default-retries", 60, "Default number of retries for various operations before panicking")
 	flag.BoolVar(&migrationContext.PanicOnWarnings, "panic-on-warnings", false, "Panic when SQL warnings are encountered when copying a batch indicating data loss")
@@ -355,6 +356,7 @@ func main() {
 	migrationContext.SetHeartbeatIntervalMilliseconds(*heartbeatIntervalMillis)
 	migrationContext.SetNiceRatio(*niceRatio)
 	migrationContext.SetChunkSize(*chunkSize)
+	migrationContext.SetChunkConcurrentSize(*chunkConcurrentSize)
 	migrationContext.SetDMLBatchSize(*dmlBatchSize)
 	migrationContext.SetMaxLagMillisecondsThrottleThreshold(*maxLagMillis)
 	migrationContext.SetThrottleQuery(*throttleQuery)

go/logic/applier.go

@@ -115,6 +115,11 @@ func (apl *Applier) InitDBConnections() (err error) {
 	if apl.db, _, err = mysql.GetDB(apl.migrationContext.Uuid, uriWithMulti); err != nil {
 		return err
 	}
+	concurrentSize := atomic.LoadInt64(&apl.migrationContext.ChunkConcurrentSize)
+	if concurrentSize > int64(mysql.MaxDBPoolConnections) {
+		apl.db.SetMaxOpenConns(int(concurrentSize) + mysql.MaxDBPoolConnections)
+		apl.db.SetMaxIdleConns(int(concurrentSize) + mysql.MaxDBPoolConnections)
+	}
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if apl.singletonDB, _, err = mysql.GetDB(apl.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err
@@ -870,10 +875,38 @@ func (apl *Applier) ReadMigrationRangeValues() error {
 }
 
 // CalculateNextIterationRangeEndValues reads the next-iteration-range-end unique key values,
-// which will be used for copying the next chunk of rows. Ir returns "false" if there is
-// no further chunk to work through, i.e. we're past the last chunk and are done with
-// iterating the range (and thus done with copying row chunks)
-func (apl *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, err error) {
+// which will be used for copying the next chunk of rows. It returns an IterationRangeValues
+// struct with HasFurtherRange=false if there is no further chunk to work through.
+// Thread-safe: uses a mutex to serialize access for concurrent row-copy.
+// When advanceCursor is true, the function determines min from MigrationIterationRangeMaxValues
+// (for concurrent mode where each goroutine advances the cursor).
+// When advanceCursor is false, min is read from MigrationIterationRangeMinValues (pre-set by
+// SetNextIterationRangeMinValues for single-threaded retry compatibility).
+func (apl *Applier) CalculateNextIterationRangeEndValues(advanceCursor bool) (values *base.IterationRangeValues, err error) {
+	apl.migrationContext.CalculateNextIterationRangeEndValuesLock.Lock()
+	defer apl.migrationContext.CalculateNextIterationRangeEndValuesLock.Unlock()
+
+	result := &base.IterationRangeValues{
+		Size: atomic.LoadInt64(&apl.migrationContext.ChunkSize),
+	}
+
+	if advanceCursor {
+		// Concurrent mode: advance min from current max cursor
+		result.Min = apl.migrationContext.MigrationIterationRangeMaxValues
+		if result.Min == nil {
+			result.Min = apl.migrationContext.MigrationRangeMinValues
+			result.IncludeMinValues = true
+		}
+	} else {
+		// Single-threaded mode: min was pre-set by SetNextIterationRangeMinValues
+		result.Min = apl.migrationContext.MigrationIterationRangeMinValues
+		if result.Min == nil {
+			result.Min = apl.migrationContext.MigrationRangeMinValues
+		}
+		// First iteration: include the minimum values (no previous max exists yet)
+		result.IncludeMinValues = (apl.migrationContext.MigrationIterationRangeMaxValues == nil)
+	}
+
 	for i := 0; i < 2; i++ {
 		buildFunc := sql.BuildUniqueKeyRangeEndPreparedQueryViaOffset
 		if i == 1 {
@@ -883,46 +916,49 @@ func (apl *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool
 			apl.migrationContext.DatabaseName,
 			apl.migrationContext.OriginalTableName,
 			&apl.migrationContext.UniqueKey.Columns,
-			apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
+			result.Min.AbstractValues(),
 			apl.migrationContext.MigrationRangeMaxValues.AbstractValues(),
-			atomic.LoadInt64(&apl.migrationContext.ChunkSize),
-			apl.migrationContext.GetIteration() == 0,
+			result.Size,
+			result.IncludeMinValues,
 			fmt.Sprintf("iteration:%d", apl.migrationContext.GetIteration()),
 		)
 		if err != nil {
-			return hasFurtherRange, err
+			return result, err
 		}
 
 		rows, err := apl.db.Query(query, explodedArgs...)
 		if err != nil {
-			return hasFurtherRange, err
+			return result, err
 		}
 		defer rows.Close()
 
 		iterationRangeMaxValues := sql.NewColumnValues(apl.migrationContext.UniqueKey.Len())
 		for rows.Next() {
 			if err = rows.Scan(iterationRangeMaxValues.ValuesPointers...); err != nil {
-				return hasFurtherRange, err
+				return result, err
 			}
-			hasFurtherRange = true
+			result.HasFurtherRange = true
 		}
 		if err = rows.Err(); err != nil {
-			return hasFurtherRange, err
+			return result, err
 		}
-		if hasFurtherRange {
-			apl.migrationContext.MigrationIterationRangeMaxValues = iterationRangeMaxValues
-			return hasFurtherRange, nil
+		if result.HasFurtherRange {
+			result.Max = iterationRangeMaxValues
+			// Advance global cursor
+			apl.migrationContext.MigrationIterationRangeMinValues = result.Min
+			apl.migrationContext.MigrationIterationRangeMaxValues = result.Max
+			return result, nil
 		}
 	}
 	apl.migrationContext.Log.Debugf("Iteration complete: no further range to iterate")
-	return hasFurtherRange, nil
+	return result, nil
 }
 
 // ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where
 // data actually gets copied from original table.
-func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
+func (apl *Applier) ApplyIterationInsertQuery(iterationRangeValues *base.IterationRangeValues) (chunkSize int64, rowsAffected int64, duration time.Duration, warnings []string, err error) {
 	startTime := time.Now()
-	chunkSize = atomic.LoadInt64(&apl.migrationContext.ChunkSize)
+	chunkSize = iterationRangeValues.Size
 
 	query, explodedArgs, err := sql.BuildRangeInsertPreparedQuery(
 		apl.migrationContext.DatabaseName,
@@ -932,52 +968,52 @@ func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected i
 		apl.migrationContext.MappedSharedColumns.Names(),
 		apl.migrationContext.UniqueKey.Name,
 		&apl.migrationContext.UniqueKey.Columns,
-		apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
-		apl.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
-		apl.migrationContext.GetIteration() == 0,
+		iterationRangeValues.Min.AbstractValues(),
+		iterationRangeValues.Max.AbstractValues(),
+		iterationRangeValues.IncludeMinValues,
 		apl.migrationContext.IsTransactionalTable(),
 		// TODO: Don't hardcode this
 		strings.HasPrefix(apl.migrationContext.ApplierMySQLVersion, "8."),
 	)
 	if err != nil {
-		return chunkSize, rowsAffected, duration, err
+		return chunkSize, rowsAffected, duration, nil, err
 	}
 
-	sqlResult, err := func() (gosql.Result, error) {
+	sqlResult, sqlWarnings, err := func() (gosql.Result, []string, error) {
 		tx, err := apl.db.Begin()
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		defer tx.Rollback()
 
 		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s'`, apl.migrationContext.ApplierTimeZone)
 		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, apl.generateSqlModeQuery())
 
 		if _, err := tx.Exec(sessionQuery); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		result, err := tx.Exec(query, explodedArgs...)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 
+		var collectedWarnings []string
 		if apl.migrationContext.PanicOnWarnings {
 			rows, err := tx.Query("SHOW WARNINGS")
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 			defer rows.Close()
 			if err = rows.Err(); err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 
 			// Compile regex once before loop to avoid performance penalty and handle errors properly
 			migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 
-			var sqlWarnings []string
 			for rows.Next() {
 				var level, message string
 				var code int
@@ -988,29 +1024,29 @@ func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected i
 				if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
 					continue
 				}
-				sqlWarnings = append(sqlWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
+				collectedWarnings = append(collectedWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
 			}
-			apl.migrationContext.MigrationLastInsertSQLWarnings = sqlWarnings
 		}
 
 		if err := tx.Commit(); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
-		return result, nil
+		return result, collectedWarnings, nil
 	}()
 
 	if err != nil {
-		return chunkSize, rowsAffected, duration, err
+		return chunkSize, rowsAffected, duration, nil, err
 	}
 	rowsAffected, _ = sqlResult.RowsAffected()
 	duration = time.Since(startTime)
+	warnings = sqlWarnings
 	apl.migrationContext.Log.Debugf(
 		"Issued INSERT on range: [%s]..[%s]; iteration: %d; chunk-size: %d",
-		apl.migrationContext.MigrationIterationRangeMinValues,
-		apl.migrationContext.MigrationIterationRangeMaxValues,
+		iterationRangeValues.Min,
+		iterationRangeValues.Max,
 		apl.migrationContext.GetIteration(),
 		chunkSize)
-	return chunkSize, rowsAffected, duration, nil
+	return chunkSize, rowsAffected, duration, warnings, nil
 }
 
 // LockOriginalTable places a write lock on the original table