Speed up parallel row-copy: overlap range calc, drop per-batch barrier, cut per-chunk round-trips

The --chunk-concurrent-size parallel row-copy only ran the INSERTs in
parallel; the boundary calculation and the per-chunk transaction overhead
serialized work and capped the achievable speedup well below the hardware's
parallel-insert ceiling. This addresses three of those caps.

Prefetch range producer (overlap serialized boundary calc with INSERTs):
- A single dedicated producer goroutine is the sole caller of
  CalculateNextIterationRangeEndValues and streams pre-computed ranges into a
  buffered channel, so boundary scans now overlap the parallel INSERTs of
  earlier work instead of stalling between batches.
- Split iterateChunks into iterateChunksSingle (unchanged single-threaded
  semantics) and iterateChunksConcurrent.
- Size the applier pool for concurrentSize + producer + headroom.

#1 Per-chunk round-trips (applier.go):
- ApplyIterationInsertQuery sent BEGIN / SET SESSION / INSERT / COMMIT as four
  round-trips per chunk. It now sends "SET SESSION ...; INSERT ..." as a single
  autocommit, multi-statement round-trip on one pinned connection. The applier
  pool already enables multiStatements + interpolateParams + autocommit;
  RowsAffected() reports the INSERT (last statement), and the optional
  SHOW WARNINGS runs on the same pinned connection. 4 round-trips -> 1.

#2 Persistent worker pool (migrator.go):
- Replace the per-batch errgroup+g.Wait barrier (which stalled N workers on
  the slowest chunk every N chunks) with continuous dispatch to an errgroup
  bounded by SetLimit(concurrentSize) for a 200ms time quantum. Workers stay
  saturated; the only barrier is at the quantum boundary. The time bound keeps
  executeWriteFuncs returning to apply binlog events and re-check throttling,
  preserving row-copy/event mutual exclusion.

Checkpoints record the last contiguous completed range (not the producer's
prefetched cursor), so resume restarts from fully-copied data.

Benchmarked on MySQL 8.0.46 (innodb_autoinc_lock_mode=2), 2.1M rows: copy time
vs the prior parallel impl improved up to 32% (chunk=200, conc=4: 22s->15s;
chunk=1000, conc=8: 8s->6s). Data integrity verified by row count + checksum.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dnovitski

doc/command-line-flags.md

@@ -84,7 +84,9 @@ See also: [`resuming-migrations`](resume.md)
 
 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.
+Each concurrent chunk calculates its own non-overlapping key range under a serialization lock, so there is no risk of duplicate or overlapping copies. A single dedicated producer goroutine streams these pre-calculated ranges to a pool of copy workers that run continuously (rather than in fixed barrier-synchronized batches), so the serialized boundary calculation overlaps with the parallel `INSERT`s and a slow chunk does not stall the others. Each chunk also applies its session variables and `INSERT` in a single autocommit round-trip, avoiding the per-chunk `BEGIN`/`SET SESSION`/`COMMIT` overhead. The applier connection pool is sized to `chunk-concurrent-size + 1 (producer) + headroom` automatically.
+
+For the speedup to materialize, MySQL should allow concurrent inserts to scale: on MySQL 8.0+ the default `innodb_autoinc_lock_mode = 2` (interleaved) is required for tables with an `AUTO_INCREMENT` column — under mode 0/1 an `INSERT ... SELECT` holds a table-level AUTO-INC lock that serializes concurrent chunks.
 
 Note: concurrency multiplies write pressure per time slot. Throttling (`--max-load`, `--nice-ratio`) applies per batch, not per chunk. Start with small values (2-8) and monitor replication lag.
 

go/logic/applier.go

@@ -116,9 +116,14 @@ func (apl *Applier) InitDBConnections() (err error) {
 		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)
+	if concurrentSize > 1 {
+		// Size the pool for concurrentSize parallel chunk-INSERTs plus the dedicated
+		// range-producer connection, with MaxDBPoolConnections of additional headroom
+		// for other applier queries. Without this, small concurrency values (2, 3)
+		// would contend with the producer for a connection and serialize.
+		poolSize := int(concurrentSize) + 1 + mysql.MaxDBPoolConnections
+		apl.db.SetMaxOpenConns(poolSize)
+		apl.db.SetMaxIdleConns(poolSize)
 	}
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if apl.singletonDB, _, err = mysql.GetDB(apl.migrationContext.Uuid, singletonApplierUri); err != nil {
@@ -958,6 +963,13 @@ func (apl *Applier) CalculateNextIterationRangeEndValues(advanceCursor bool) (va
 
 // ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where
 // data actually gets copied from original table.
+//
+// The session variables (time_zone, sql_mode) and the chunk INSERT are sent as a single
+// multi-statement, autocommit round-trip on one pinned connection. The applier pool sets
+// multiStatements + interpolateParams + autocommit, so this avoids the extra
+// BEGIN / SET SESSION / COMMIT round-trips an explicit transaction would add to every
+// chunk — the dominant per-chunk overhead at small chunk sizes. `RowsAffected()` reports
+// the last statement's count (the INSERT), so the returned row count stays correct.
 func (apl *Applier) ApplyIterationInsertQuery(iterationRangeValues *base.IterationRangeValues) (chunkSize int64, rowsAffected int64, duration time.Duration, warnings []string, err error) {
 	startTime := time.Now()
 	chunkSize = iterationRangeValues.Size
@@ -981,60 +993,31 @@ func (apl *Applier) ApplyIterationInsertQuery(iterationRangeValues *base.Iterati
 		return chunkSize, rowsAffected, duration, nil, err
 	}
 
+	sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s', %s`,
+		apl.migrationContext.ApplierTimeZone, apl.generateSqlModeQuery())
+	combinedQuery := sessionQuery + "; " + query
+
+	ctx := apl.migrationContext.GetContext()
 	sqlResult, sqlWarnings, err := func() (gosql.Result, []string, error) {
-		tx, err := apl.db.Begin()
+		// Pin a single connection so the optional SHOW WARNINGS observes this INSERT's
+		// warnings (and not another pooled query's).
+		conn, err := apl.db.Conn(ctx)
 		if err != nil {
 			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())
+		defer conn.Close()
 
-		if _, err := tx.Exec(sessionQuery); err != nil {
-			return nil, nil, err
-		}
-		result, err := tx.Exec(query, explodedArgs...)
+		result, err := conn.ExecContext(ctx, combinedQuery, explodedArgs...)
 		if err != nil {
 			return nil, nil, err
 		}
 
 		var collectedWarnings []string
 		if apl.migrationContext.PanicOnWarnings {
-			rows, err := tx.Query("SHOW WARNINGS")
-			if err != nil {
-				return nil, nil, err
-			}
-			defer rows.Close()
-			if err = rows.Err(); err != nil {
-				return nil, nil, err
-			}
-
-			// Compile regex once before loop to avoid performance penalty and handle errors properly
-			migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
+			collectedWarnings, err = apl.collectChunkInsertWarnings(ctx, conn)
 			if err != nil {
 				return nil, nil, err
 			}
-
-			for rows.Next() {
-				var level, message string
-				var code int
-				if err := rows.Scan(&level, &code, &message); err != nil {
-					apl.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
-					continue
-				}
-				if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
-					continue
-				}
-				collectedWarnings = append(collectedWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
-			}
-			if err := rows.Err(); err != nil {
-				return nil, nil, err
-			}
-		}
-
-		if err := tx.Commit(); err != nil {
-			return nil, nil, err
 		}
 		return result, collectedWarnings, nil
 	}()
@@ -1054,6 +1037,41 @@ func (apl *Applier) ApplyIterationInsertQuery(iterationRangeValues *base.Iterati
 	return chunkSize, rowsAffected, duration, warnings, nil
 }
 
+// collectChunkInsertWarnings runs SHOW WARNINGS on the given (pinned) connection right
+// after a chunk INSERT and returns the warnings, skipping the benign duplicate-key
+// warnings that INSERT IGNORE produces on the migration unique key.
+func (apl *Applier) collectChunkInsertWarnings(ctx context.Context, conn *gosql.Conn) ([]string, error) {
+	rows, err := conn.QueryContext(ctx, "SHOW WARNINGS")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	// Compile regex once before loop to avoid performance penalty and handle errors properly
+	migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
+	if err != nil {
+		return nil, err
+	}
+
+	var collectedWarnings []string
+	for rows.Next() {
+		var level, message string
+		var code int
+		if err := rows.Scan(&level, &code, &message); err != nil {
+			apl.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
+			continue
+		}
+		if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
+			continue
+		}
+		collectedWarnings = append(collectedWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return collectedWarnings, nil
+}
+
 // LockOriginalTable places a write lock on the original table
 func (apl *Applier) LockOriginalTable() error {
 	query := fmt.Sprintf(`lock /* gh-ost */ tables %s.%s write`,