corrections after review and testing

Author: olegkv

doc/command-line-flags.md

@@ -110,10 +110,10 @@ Controls how many rows `gh-ost` copies in each row-copy iteration. The default i
 
 Parallel copy requires checkpoints because parallel INSERTs can go out of sequential order and leave gaps in target table. gh-ost checkpoints are consistent and do not have gaps, so resuming from a checkpoint will make the target table consistent again.
 
-Consistency is ensured by using 
--INSERT IGNORE/SELECT FOR SHARE for row copy (current gh-ost behavior)
--DELETE/INSERT by DML applier when unique key change is detected (current gh-ost behavior)
--advanceFrontier call to ensure that checkpoints have no gaps (used by parallel-copy only)
+Consistency is ensured by using:
+ - `INSERT IGNORE`/`SELECT FOR SHARE` for row copy (current gh-ost behavior)
+ - `DELETE`/`INSERT` by DML applier when unique key change is detected (current gh-ost behavior)
+ - `advanceFrontier` call to ensure that checkpoints have no gaps (used by parallel-copy only)
 Without all of those, consistency of parallel copy cannot be guaranteed (for example, row copy could insert row back after it was deleted). In existing serial mode, the problem is much easier because DML applier is never parallel with row copy.
 
 ### parallel-copy-workers

go/base/context.go

@@ -56,16 +56,16 @@ const (
 	MaxEventsBatchSize = 1000
 	ETAUnknown         = math.MinInt64
 
-	// MaxCopyWorkers is the hard upper bound on --parallel-copy-workers. The applier
-	// connection pool is sized to CopyWorkers + ParallelCopyConnHeadroom, so this
-	// also bounds connection usage and prevents exhausting the server's
+	// MaxParallelCopyWorkers is the hard upper bound on --parallel-copy-workers. The
+	// applier connection pool is sized to ParallelCopyWorkers + ParallelCopyConnHeadroom,
+	// so this also bounds connection usage and prevents exhausting the server's
 	// max_connections.
-	MaxCopyWorkers = 64
-	// RecommendedMaxCopyWorkers is a soft threshold above which we warn: the
+	MaxParallelCopyWorkers = 64
+	// RecommendedMaxParallelCopyWorkers is a soft threshold above which we warn: the
 	// throughput benefit of more workers plateaus while load keeps growing.
-	RecommendedMaxCopyWorkers = 16
+	RecommendedMaxParallelCopyWorkers = 16
 	// ParallelCopyConnHeadroom is the number of connections added to the applier
-	// pool on top of CopyWorkers, to cover the DML event applier goroutine
+	// pool on top of ParallelCopyWorkers, to cover the DML event applier goroutine,
 	// the status logger, the heartbeat goroutine, and the checkpoint loop.
 	ParallelCopyConnHeadroom = 4
 )

go/cmd/gh-ost/main.go

@@ -347,13 +347,13 @@ func main() {
 		if migrationContext.ParallelCopyWorkers < 1 {
 			migrationContext.Log.Fatalf("--parallel-copy-workers should be >=1")
 		}
-		if migrationContext.ParallelCopyWorkers > base.MaxCopyWorkers {
-			migrationContext.Log.Fatalf("--parallel-copy-workers should be <=%d", base.MaxCopyWorkers)
+		if migrationContext.ParallelCopyWorkers > base.MaxParallelCopyWorkers {
+			migrationContext.Log.Fatalf("--parallel-copy-workers should be <=%d", base.MaxParallelCopyWorkers)
 		}
-		if migrationContext.ParallelCopyWorkers > base.RecommendedMaxCopyWorkers {
-			migrationContext.Log.Warningf("--parallel-copy-workers=%d is high: throughput gains tend to plateau while load on the server grows. Consider <=%d.", migrationContext.ParallelCopyWorkers, base.RecommendedMaxCopyWorkers)
+		if migrationContext.ParallelCopyWorkers > base.RecommendedMaxParallelCopyWorkers {
+			migrationContext.Log.Warningf("--parallel-copy-workers=%d is high: throughput gains tend to plateau while load on the server grows. Consider <=%d.", migrationContext.ParallelCopyWorkers, base.RecommendedMaxParallelCopyWorkers)
 		}
-		if migrationContext.ParallelCopyMaxHeartbeatLagThresholdMillies > 0 && migrationContext.ParallelCopyMaxHeartbeatLagThresholdMillies < 100 {
+		if migrationContext.ParallelCopyMaxHeartbeatLagThresholdMillies != 0 && migrationContext.ParallelCopyMaxHeartbeatLagThresholdMillies < 100 {
 			migrationContext.Log.Fatalf("--parallel-copy-max-heartbeatlag-millis should be >=100 or 0 (disabled)")
 		}
 	}

go/logic/applier.go

@@ -1178,7 +1178,7 @@ func (apl *Applier) ApplyIterationInsertQuery(parallelRange ...*iterationRange)
 	if parallel {
 		// Use the explicitly-passed range. The parallel INSERT runs unlocked and
 		// concurrently, so it must NOT read the shared MigrationIterationRange* cursor:
-		// another worker is mutating it under parallelSelectMutex.
+		// another worker is mutating it under parallelCopySelectMutex.
 		rangeMin = parallelRange[0].min
 		rangeMax = parallelRange[0].max
 		includeRangeStartValues = parallelRange[0].includeRangeStart

go/logic/migrator.go

@@ -1520,7 +1520,7 @@ func (mgtr *Migrator) printStatus(rule PrintStatusRule, snap migrationProgressSn
 
 	throughputSuffix := ""
 	if mgtr.migrationContext.ParallelCopy {
-		throughputSuffix = fmt.Sprintf("; Throughput: %drows/sec", atomic.LoadInt64(&mgtr.migrationContext.EtaRowsPerSecond))
+		throughputSuffix = fmt.Sprintf("; Throughput: %d rows/sec", atomic.LoadInt64(&mgtr.migrationContext.EtaRowsPerSecond))
 	}
 	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, HeartbeatLag: %.2fs, State: %s; ETA: %s%s",
 		snap.totalRowsCopied, snap.rowsEstimate, snap.progressPct,
@@ -1708,7 +1708,7 @@ func (mgtr *Migrator) iterateChunks() error {
 			return nil
 		}
 		// --parallel-copy state, captured once per copyRowsFunc invocation under
-		// parallelSelectMutex so the chunk INSERT (which runs unlocked, in parallel) uses
+		// parallelCopySelectMutex so the chunk INSERT (which runs unlocked, in parallel) uses
 		// a stable range even as other workers advance the shared cursor, and so a retry
 		// re-inserts the same chunk rather than re-scanning a new range.
 		var parallelIteration int64
@@ -1720,10 +1720,14 @@ func (mgtr *Migrator) iterateChunks() error {
 			}
 			// Copy task:
 			applyCopyRowsFunc := func() error {
-				if atomic.LoadInt64(&mgtr.rowCopyCompleteFlag) == 1 || atomic.LoadInt64(&hasNoFurtherRangeFlag) == 1 {
-					// Done.
-					// There's another such check down the line
-					return nil
+				// Once this closure has captured a chunk range, it must keep retrying that
+				// same INSERT to completion even if hasNoFurtherRangeFlag/rowCopyCompleteFlag is set
+				if !parallelCaptured {
+					if atomic.LoadInt64(&mgtr.rowCopyCompleteFlag) == 1 || atomic.LoadInt64(&hasNoFurtherRangeFlag) == 1 {
+						// Done.
+						// There's another such check down the line
+						return nil
+					}
 				}
 
 				if mgtr.migrationContext.ParallelCopy {
@@ -1733,7 +1737,7 @@ func (mgtr *Migrator) iterateChunks() error {
 					// advanceFrontier is never left with an abandoned iteration.
 					if !parallelCaptured {
 						var stopErr error
-						parallelIteration, parallelRangeMin, parallelRangeMax, parallelCaptured, stopErr = mgtr.captureParallelChunkRange(&hasNoFurtherRangeFlag, terminateRowIteration)
+						parallelIteration, parallelRangeMin, parallelRangeMax, parallelCaptured, stopErr = mgtr.captureParallelChunkRange(&hasNoFurtherRangeFlag)
 						if !parallelCaptured {
 							return stopErr
 						}
@@ -1819,9 +1823,9 @@ func (mgtr *Migrator) iterateChunks() error {
 		}
 
 		// Enqueue copy operation. In serial mode it is consumed by executeWriteFuncs(); in
-		// --parallel-copy mode it is consumed by one of the CopyWorkers workers started in
-		// copyRowsParallel(). The closure itself serializes its boundary SELECT under
-		// parallelSelectMutex, so the only thing parallelized is the chunk INSERT.
+		// --parallel-copy mode it is consumed by one of the ParallelCopyWorkers worker
+		// goroutines started in copyRowsParallel(). The closure itself serializes its boundary SELECT under
+		// parallelCopySelectMutex, so the only thing parallelized is the chunk INSERT.
 		// Use helper to prevent deadlock if the consumer exits.
 		if err := base.SendWithContext(mgtr.migrationContext.GetContext(), mgtr.copyRowsQueue, copyRowsFunc); err != nil {
 			// Context cancelled, check for abort and exit
@@ -1833,13 +1837,14 @@ func (mgtr *Migrator) iterateChunks() error {
 	}
 }
 
-// captureParallelChunkRange acquires parallelSelectMutex, advances the shared scan
-// cursor, and captures the next chunk's range. Returns captured=true with the range
-// values on success. Returns captured=false with nil err when no further work is needed
-// (row copy already complete or table exhausted; in the latter case terminateRowIteration
-// is called to signal the channel). Returns captured=false with non-nil err when the
-// range scan itself fails; the caller should propagate it to trigger a retry.
-func (mgtr *Migrator) captureParallelChunkRange(hasNoFurtherRangeFlag *int64, terminateRowIteration func(error) error) (iteration int64, rangeMin, rangeMax *sql.ColumnValues, captured bool, err error) {
+// This helper never signals clean completion on rowCopyComplete itself: that signal must
+// fire exactly once, after copyRowsParallel's wg.Wait() confirms every worker goroutine
+// has actually returned — not merely that the scan is exhausted. Signaling here, as soon
+// as no further range is found, would let the migrator start moving toward cutover while
+// other workers may still be retrying an in-flight INSERT for a chunk they already
+// captured. (That a captured chunk is never dropped once other workers see this flag is a
+// separate guarantee, enforced by applyCopyRowsFunc's parallelCaptured guard.)
+func (mgtr *Migrator) captureParallelChunkRange(hasNoFurtherRangeFlag *int64) (iteration int64, rangeMin, rangeMax *sql.ColumnValues, captured bool, err error) {
 	mgtr.parallelCopySelectMutex.Lock()
 	if atomic.LoadInt64(&mgtr.rowCopyCompleteFlag) == 1 || atomic.LoadInt64(hasNoFurtherRangeFlag) == 1 {
 		mgtr.parallelCopySelectMutex.Unlock()
@@ -1854,7 +1859,7 @@ func (mgtr *Migrator) captureParallelChunkRange(hasNoFurtherRangeFlag *int64, te
 	if !hasFurtherRange {
 		atomic.StoreInt64(hasNoFurtherRangeFlag, 1)
 		mgtr.parallelCopySelectMutex.Unlock()
-		return 0, nil, nil, false, terminateRowIteration(nil)
+		return 0, nil, nil, false, nil
 	}
 	iteration = atomic.LoadInt64(&mgtr.parallelCopyDispatchSeq)
 	rangeMin = mgtr.migrationContext.MigrationIterationRangeMinValues.Clone()

go/logic/parallel_copy.go

@@ -57,8 +57,8 @@ func (mgtr *Migrator) throttleOnHeartbeatLag(ctx context.Context) {
 
 // copyRowsParallel is the --parallel-copy consumer side. iterateChunks remains the single
 // producer, enqueuing copy-task closures onto copyRowsQueue exactly as in serial mode; this
-// function starts CopyWorkers consumer goroutines that pull those closures and run them
-// concurrently. Each closure serializes its own boundary SELECT under parallelSelectMutex
+// function starts ParallelCopyWorkers consumer goroutines that pull those closures and run them
+// concurrently. Each closure serializes its own boundary SELECT under parallelCopySelectMutex
 // (so chunk ranges are still produced sequentially) and only the chunk INSERT runs in
 // parallel; advanceFrontier then commits global state in contiguous iteration order so a
 // crash/resume with checkpoints never leaves un-copied holes.

localtests/parallel-copy-resume/create.sql

@@ -17,7 +17,7 @@ insert into gh_ost_test (i, color) values
 -- run spends well over one checkpoint interval (10s) in row-copy. This makes it
 -- likely that a checkpoint lands mid-copy, so --resume exercises gap-filling
 -- from a partial frontier. (The test stays correct either way; the deterministic
--- gap logic is covered by go/logic/parallel_test.go.)
+-- gap logic is covered by go/logic/parallel_copy_test.go.)
 insert into gh_ost_test (i, color) select i, color from gh_ost_test;
 insert into gh_ost_test (i, color) select i, color from gh_ost_test;
 insert into gh_ost_test (i, color) select i, color from gh_ost_test;

localtests/parallel-copy-resume/test.sh

@@ -6,7 +6,7 @@
 # The test is robust to timing: if the first run happens to finish the row copy
 # before we kill it, --resume simply has no gaps to fill and completes. Either
 # way the ghost table must match the original. The deterministic gap-filling
-# logic itself is covered by the unit tests in go/logic/parallel_test.go.
+# logic itself is covered by the unit tests in go/logic/parallel_copy_test.go.
 
 table_name="gh_ost_test"
 ghost_table_name="_gh_ost_test_gho"