Parallel row-copy, DML merge, frontier filter, and heartbeat lag throttle (#2)

Performance optimizations for gh-ost that significantly speed up row-copy
under high write load while keeping binlog lag bounded:

- Parallel row-copy with dedicated connection pool and time-bounded drain
- DML event merging within batches (INSERT/DELETE cancellation, UPDATE folding)
- Frontier filter to skip DML events beyond copy frontier
- Heartbeat lag throttle (--copy-max-lag-millis) for row-copy pacing
- Adaptive drain budget and auto-tuning chunk size
- Runtime-changeable --copy-concurrency and --copy-max-lag-millis
- Fix multithreaded replication data inconsistency

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

Author: dnovitski

.gitignore

@@ -4,3 +4,4 @@
 /.vendor/
 .idea/
 *.tmp
+gh-ost

doc/command-line-flags.md

@@ -92,6 +92,30 @@ password=123456
 
 Defaults to `true`. See [`exact-rowcount`](#exact-rowcount)
 
+### copy-concurrency
+
+Number of concurrent row-copy goroutines. Default: `1` (original single-threaded behavior). Range: `1-32`.
+
+Higher values speed up row-copy under heavy write load by using parallel `INSERT...SELECT` operations on non-overlapping ranges. Each worker copies a different chunk concurrently, and a bounded drain budget ensures DML events are still processed between copy batches.
+
+When `--copy-concurrency` > 1, the [heartbeat lag throttle](#copy-max-lag-millis) is automatically active to prevent unbounded binlog lag growth.
+
+This value can be changed at runtime via [interactive commands](interactive-commands.md).
+
+Example: `--copy-concurrency=4` — uses 4 parallel copy workers.
+
+### copy-max-lag-millis
+
+Maximum allowed *internal* binlog processing lag (HeartbeatLag) in milliseconds before row-copy is paused. Default: `60000` (60 seconds). Set to `0` to disable.
+
+This is only active when `--copy-concurrency` > 1. When HeartbeatLag exceeds this threshold, row-copy pauses and gh-ost drains the binlog event queue exclusively until lag drops to half the threshold (hysteresis prevents oscillation).
+
+**Differs from [`--max-lag-millis`](#max-lag-millis):** `--max-lag-millis` measures *replica* replication lag and throttles the entire migration. `--copy-max-lag-millis` measures gh-ost's *own* binlog processing lag and only pauses row-copy while DML apply continues at full speed.
+
+This value can be changed at runtime via [interactive commands](interactive-commands.md).
+
+See [Throttle: Copy-specific lag throttle](throttle.md#copy-specific-lag-throttle---copy-max-lag-millis) for a detailed comparison.
+
 ### critical-load
 
 Comma delimited status-name=threshold, same format as [`--max-load`](#max-load).
@@ -200,6 +224,16 @@ See [`initially-drop-ghost-table`](#initially-drop-ghost-table)
 
 Default False. Should `gh-ost` forcibly delete an existing socket file. Be careful: this might drop the socket file of a running migration!
 
+### copy-max-lag-millis
+
+When using parallel row-copy (`--copy-concurrency` > 1), this flag sets the maximum allowed *internal* binlog processing lag (HeartbeatLag) before row-copy is paused. Unlike `--max-lag-millis` which measures replica lag and throttles the entire migration, `--copy-max-lag-millis` only pauses row-copy while DML event processing continues at full speed.
+
+Default: `60000` (60 seconds). Set to `0` to disable (maximum copy speed, unbounded lag).
+
+Row-copy resumes when HeartbeatLag drops to half the threshold (hysteresis prevents oscillation).
+
+See [Throttle: Copy-specific lag throttle](throttle.md#copy-specific-lag-throttle---copy-max-lag-millis) for a detailed comparison with `--max-lag-millis`.
+
 ### max-lag-millis
 
 On a replication topology, this is perhaps the most important migration throttling factor: the maximum lag allowed for migration to work. If lag exceeds this value, migration throttles.
@@ -252,6 +286,28 @@ Defaults to an auto-determined and advertised upon startup file. Defines Unix so
 
 By default `gh-ost` verifies no foreign keys exist on the migrated table. On servers with large number of tables this check can take a long time. If you're absolutely certain no foreign keys exist (table does not reference other table nor is referenced by other tables) and wish to save the check time, provide with `--skip-foreign-key-checks`.
 
+### skip-dml-frontier-filter
+
+Disable the frontier-based DML skip optimization. Default: `false` (optimization is enabled).
+
+When enabled (default), gh-ost skips DML events targeting rows that have not yet been copied — since row-copy will capture the latest value when it reaches those rows. This reduces redundant DML apply work during the copy phase.
+
+**Safety constraints** — the frontier filter is automatically disabled in two situations regardless of this flag:
+
+1. **When `--copy-concurrency` > 1**: With parallel row-copy, multiple chunks are being copied concurrently. The frontier (last completed chunk boundary) is not a reliable boundary because in-flight chunks may not have committed yet. A DML event targeting a row in an in-flight chunk could be incorrectly skipped, causing data loss.
+
+2. **In replica modes** (`--test-on-replica`, `--migrate-on-replica`): In replica mode, gh-ost reads binlog events from the replica's relay log. These events may be ahead of the replica's SQL thread apply position — meaning row-copy `SELECT` queries may not yet see the data from the skipped events. This would cause silent data loss since neither the DML apply nor the row-copy would capture those changes.
+
+Use `--skip-dml-frontier-filter` for benchmarking or if you suspect the optimization is causing issues.
+
+### skip-dml-merge
+
+Disable DML event merging within batches. Default: `false` (merging is enabled).
+
+When enabled (default), gh-ost merges redundant DML events for the same row before applying them. For example, an INSERT followed by multiple UPDATEs to the same row becomes a single INSERT with the final values. Under high write load, this can reduce applied statements by ~36%.
+
+Use `--skip-dml-merge` for benchmarking or if you suspect merging is causing issues.
+
 ### skip-metadata-lock-check
 
 By default `gh-ost` performs a check before the cut-over to ensure the rename session holds the exclusive metadata lock on the table. In case `performance_schema.metadata_locks` cannot be enabled on your setup, this check can be skipped with `--skip-metadata-lock-check`. 
@@ -333,3 +389,11 @@ Makes the _old_ table include a timestamp value. The _old_ table is what the ori
 ### tungsten
 
 See [`tungsten`](cheatsheet.md#tungsten) on the cheatsheet.
+
+### workers
+
+Number of concurrent workers for applying DML events. Default: `8`. Each worker uses one goroutine and its own database connection to apply binlog events in parallel.
+
+This controls the parallelism of the multithreaded replication (MTR) subsystem which processes binlog DML events. Higher values can improve DML throughput under heavy write load, but beyond 8-16 workers the gains diminish as contention on the ghost table increases.
+
+Example: `--workers=16` — uses 16 parallel DML apply workers.

doc/interactive-commands.md

@@ -24,6 +24,8 @@ Both interfaces may serve at the same time. Both respond to simple text command,
 - `chunk-size=<newsize>`: modify the `chunk-size`; applies on next running copy-iteration
 - `dml-batch-size=<newsize>`: modify the `dml-batch-size`; applies on next applying of binary log events
 - `max-lag-millis=<max-lag>`: modify the maximum replication lag threshold (milliseconds, minimum value is `100`, i.e. `0.1` second)
+- `copy-concurrency=<N>`: modify the number of parallel row-copy workers (range 1-32). Setting to `1` switches to single-threaded legacy behavior. Takes effect on the next copy iteration.
+- `copy-max-lag-millis=<millis>`: modify the heartbeat lag threshold for copy throttling (milliseconds). Set `0` to disable. Takes effect immediately on the next drain check.
 - `max-load=<max-load-thresholds>`: modify the `max-load` config; applies on next running copy-iteration
   - The `max-load` format must be: `some_status=<numeric-threshold>[,some_status=<numeric-threshold>...]`'
   - For example: `Threads_running=50,threads_connected=1000`, and you would then write/echo `max-load=Threads_running=50,threads_connected=1000` to the socket.

doc/throttle.md

@@ -76,6 +76,26 @@ In addition to the above, you are able to take control and throttle the operatio
     echo no-throttle | nc -U /tmp/gh-ost.test.sample_data_0.sock
   ```
 
+### Copy-specific lag throttle (--copy-max-lag-millis)
+
+When using parallel row-copy (`--copy-concurrency` > 1), gh-ost uses a bounded drain strategy that gives row-copy more execution turns. This can cause the binlog processing to fall behind, resulting in increasing "HeartbeatLag" — the time since the last heartbeat event was *processed* from the binlog stream.
+
+`--copy-max-lag-millis` (default: `60000`, i.e. 60 seconds) sets a threshold for how far behind binlog processing may fall during the row-copy phase. When HeartbeatLag exceeds this threshold, row-copy is paused and gh-ost drains the binlog event queue exclusively until lag drops to half the threshold (hysteresis to prevent oscillation).
+
+**How it differs from `--max-lag-millis`:**
+
+| | `--max-lag-millis` | `--copy-max-lag-millis` |
+|---|---|---|
+| **What it measures** | Replication lag on downstream replicas (via heartbeat table queries or `SHOW SLAVE STATUS`) | Binlog processing lag *within gh-ost itself* (wall-clock time since last heartbeat event was read from the binlog stream) |
+| **What it throttles** | The entire migration — all reads and writes pause | Only row-copy — DML event processing continues at full speed |
+| **When it matters** | Always — protects replicas from falling behind | Only with parallel row-copy (`--copy-concurrency` > 1) |
+| **Default** | 1500ms (strict — sub-second replica lag) | 60000ms (permissive — allows some internal lag buildup for speed) |
+| **Resume behavior** | Resumes immediately when lag drops below threshold | Resumes when lag drops to threshold/2 (hysteresis) |
+
+**Why both are needed:** `--max-lag-millis` protects *downstream replicas* from excessive replication lag. `--copy-max-lag-millis` protects against *gh-ost's own internal lag* — which only occurs with parallel row-copy because the bounded drain budget gives copy more turns at the expense of binlog processing. With single-copy mode (the default), `ProcessEventsUntilDrained()` blocks indefinitely so internal lag never accumulates.
+
+Set `--copy-max-lag-millis=0` to disable the throttle entirely (maximum copy speed, unbounded lag growth).
+
 ### Throttle precedence
 
 Any single factor in the above that suggests the migration should throttle - causes throttling. That is, once some component decides to throttle, you cannot override it; you cannot force continued execution of the migration.

go/base/context.go

@@ -176,6 +176,23 @@ type MigrationContext struct {
 	CutOverType                  CutOver
 	ReplicaServerId              uint
 
+	// Number of workers used by the trx coordinator
+	NumWorkers int
+
+	// Number of concurrent row-copy goroutines (default 1 = legacy behavior)
+	CopyConcurrency int64
+
+	// SkipDMLMerge disables DML event merging within batches
+	SkipDMLMerge bool
+
+	// SkipDMLFrontierFilter disables the frontier-based DML skip optimization
+	SkipDMLFrontierFilter bool
+
+	// CopyMaxLagMillis is the heartbeat lag threshold (in ms) at which row-copy
+	// is paused to let binlog processing catch up. Only applies when copy-concurrency > 1.
+	// Default 60000 (60s). Set to 0 to disable lag-based copy throttling.
+	CopyMaxLagMillis int64
+
 	Hostname                               string
 	AssumeMasterHostname                   string
 	ApplierTimeZone                        string

go/binlog/gomysql_reader.go

@@ -1,21 +1,19 @@
 /*
    Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
+ See https://github.com/github/gh-ost/blob/master/LICENSE
 */
 
 package binlog
 
 import (
+	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/mysql"
-	"github.com/github/gh-ost/go/sql"
 
-	"time"
-
-	"context"
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
 	uuid "github.com/google/uuid"
@@ -85,53 +83,17 @@ func (gmr *GoMySQLReader) GetCurrentBinlogCoordinates() mysql.BinlogCoordinates
 	return gmr.currentCoordinates.Clone()
 }
 
-func (gmr *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
-	currentCoords := gmr.GetCurrentBinlogCoordinates()
-	dml := ToEventDML(ev.Header.EventType.String())
-	if dml == NotDML {
-		return fmt.Errorf("unknown DML type: %s", ev.Header.EventType.String())
-	}
-	for i, row := range rowsEvent.Rows {
-		if dml == UpdateDML && i%2 == 1 {
-			// An update has two rows (WHERE+SET)
-			// We do both at the same time
-			continue
+// StreamEvents reads binlog events and sends them to the given channel.
+// It is blocking and should be executed in a goroutine.
+func (gmr *GoMySQLReader) StreamEvents(ctx context.Context, canStopStreaming func() bool, eventChannel chan<- *replication.BinlogEvent) error {
+	for {
+		if canStopStreaming() {
+			return nil
 		}
-		binlogEntry := NewBinlogEntryAt(currentCoords)
-		binlogEntry.DmlEvent = NewBinlogDMLEvent(
-			string(rowsEvent.Table.Schema),
-			string(rowsEvent.Table.Table),
-			dml,
-		)
-		switch dml {
-		case InsertDML:
-			{
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
-			}
-		case UpdateDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(rowsEvent.Rows[i+1])
-			}
-		case DeleteDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-			}
+		if err := ctx.Err(); err != nil {
+			return err
 		}
-
-		// The channel will do the throttling. Whoever is reading from the channel
-		// decides whether action is taken synchronously (meaning we wait before
-		// next iteration) or asynchronously (we keep pushing more events)
-		// In reality, reads will be synchronous
-		entriesChannel <- binlogEntry
-	}
-	return nil
-}
-
-// StreamEvents
-func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
-	for !canStopStreaming() {
-		ev, err := gmr.binlogStreamer.GetEvent(context.Background())
+		ev, err := gmr.binlogStreamer.GetEvent(ctx)
 		if err != nil {
 			return err
 		}
@@ -153,45 +115,38 @@ func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChan
 
 		switch event := ev.Event.(type) {
 		case *replication.GTIDEvent:
-			if !gmr.migrationContext.UseGTIDs {
-				continue
-			}
-			sid, err := uuid.FromBytes(event.SID)
-			if err != nil {
-				return err
-			}
-			gmr.currentCoordinatesMutex.Lock()
-			if gmr.LastTrxCoords != nil {
-				gmr.currentCoordinates = gmr.LastTrxCoords.Clone()
+			if gmr.migrationContext.UseGTIDs {
+				sid, err := uuid.FromBytes(event.SID)
+				if err != nil {
+					return err
+				}
+				gmr.currentCoordinatesMutex.Lock()
+				if gmr.LastTrxCoords != nil {
+					gmr.currentCoordinates = gmr.LastTrxCoords.Clone()
+				}
+				coords := gmr.currentCoordinates.(*mysql.GTIDBinlogCoordinates)
+				trxGset := gomysql.NewUUIDSet(sid, gomysql.Interval{Start: event.GNO, Stop: event.GNO + 1})
+				coords.GTIDSet.AddSet(trxGset)
+				gmr.currentCoordinatesMutex.Unlock()
 			}
-			coords := gmr.currentCoordinates.(*mysql.GTIDBinlogCoordinates)
-			trxGset := gomysql.NewUUIDSet(sid, gomysql.Interval{Start: event.GNO, Stop: event.GNO + 1})
-			coords.GTIDSet.AddSet(trxGset)
-			gmr.currentCoordinatesMutex.Unlock()
 		case *replication.RotateEvent:
-			if gmr.migrationContext.UseGTIDs {
-				continue
+			if !gmr.migrationContext.UseGTIDs {
+				gmr.currentCoordinatesMutex.Lock()
+				coords := gmr.currentCoordinates.(*mysql.FileBinlogCoordinates)
+				coords.LogFile = string(event.NextLogName)
+				gmr.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", coords.LogFile, int64(ev.Header.LogPos), event.NextLogName)
+				gmr.currentCoordinatesMutex.Unlock()
 			}
-			gmr.currentCoordinatesMutex.Lock()
-			coords := gmr.currentCoordinates.(*mysql.FileBinlogCoordinates)
-			coords.LogFile = string(event.NextLogName)
-			gmr.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", coords.LogFile, int64(ev.Header.LogPos), event.NextLogName)
-			gmr.currentCoordinatesMutex.Unlock()
 		case *replication.XIDEvent:
 			if gmr.migrationContext.UseGTIDs {
 				gmr.LastTrxCoords = &mysql.GTIDBinlogCoordinates{GTIDSet: event.GSet.(*gomysql.MysqlGTIDSet)}
 			} else {
 				gmr.LastTrxCoords = gmr.currentCoordinates.Clone()
 			}
-		case *replication.RowsEvent:
-			if err := gmr.handleRowsEvent(ev, event, entriesChannel); err != nil {
-				return err
-			}
 		}
-	}
-	gmr.migrationContext.Log.Debugf("done streaming events")
 
-	return nil
+		eventChannel <- ev
+	}
 }
 
 func (gmr *GoMySQLReader) Close() error {

go/cmd/gh-ost/main.go

@@ -112,6 +112,11 @@ func main() {
 	flag.BoolVar(&migrationContext.PanicOnWarnings, "panic-on-warnings", false, "Panic when SQL warnings are encountered when copying a batch indicating data loss")
 	cutOverLockTimeoutSeconds := flag.Int64("cut-over-lock-timeout-seconds", 3, "Max number of seconds to hold locks on tables while attempting to cut-over (retry attempted when lock exceeds timeout) or attempting instant DDL")
 	niceRatio := flag.Float64("nice-ratio", 0, "force being 'nice', imply sleep time per chunk time; range: [0.0..100.0]. Example values: 0 is aggressive. 1: for every 1ms spent copying rows, sleep additional 1ms (effectively doubling runtime); 0.7: for every 10ms spend in a rowcopy chunk, spend 7ms sleeping immediately after")
+	flag.IntVar(&migrationContext.NumWorkers, "workers", 8, "Number of concurrent workers for applying DML events. Each worker uses one goroutine.")
+	flag.Int64Var(&migrationContext.CopyConcurrency, "copy-concurrency", 1, "Number of concurrent row-copy goroutines. Higher values speed up row-copy under write load by using parallel INSERT...SELECT operations on non-overlapping ranges. (range 1-32)")
+	flag.BoolVar(&migrationContext.SkipDMLMerge, "skip-dml-merge", false, "Disable DML event merging within batches (for benchmarking)")
+	flag.BoolVar(&migrationContext.SkipDMLFrontierFilter, "skip-dml-frontier-filter", false, "Disable frontier-based DML skip optimization (for benchmarking)")
+	flag.Int64Var(&migrationContext.CopyMaxLagMillis, "copy-max-lag-millis", 60000, "Heartbeat lag (ms) at which row-copy pauses to let binlog catch up. Only active with --copy-concurrency > 1. Set 0 to disable. Default 60s balances copy speed vs lag growth.")
 
 	maxLagMillis := flag.Int64("max-lag-millis", 1500, "replication lag at which to throttle operation")
 	replicationLagQuery := flag.String("replication-lag-query", "", "Deprecated. gh-ost uses an internal, subsecond resolution query")