fix(applier): disable NOWAIT for row-copy under MTS to avoid 3572 FATAL

Under MTS mode (NumWorkers > 1), parallel DML workers hold row locks on
the ghost table concurrently with row-copy. SELECT ... FOR SHARE NOWAIT
fails immediately with errno 3572 when it encounters those locks, and
the combined retry budget (20 internal + 3 outer) is insufficient under
heavy write load on MySQL 8.4, causing the migration to FATAL.

Extract the noWait decision into rowCopyUsesNoWait() which returns
false when NumWorkers > 1, making row-copy use LOCK IN SHARE MODE
(blocking wait) instead. This is the correct trade-off: DML apply is
latency-sensitive and should not be preempted by row-copy retries, so
row-copy should wait for DML locks to release rather than failing fast.

Single-threaded mode on MySQL 8.x retains NOWAIT behaviour unchanged.

Author: jackiesre721

go/logic/applier.go

@@ -1107,12 +1107,24 @@ func (apl *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool
 }
 
 // rowCopyMaxTransientRetries bounds internal retries of the chunk-INSERT query on
-// transient lock errors (NOWAIT 3572, lock-wait timeout 1205, deadlock 1213).
-// Under heavy concurrent write load the SELECT ... FOR SHARE NOWAIT clause can
-// fail immediately; retrying with short backoff avoids exhausting the outer
-// retryOperation budget (which defaults to only 3 attempts).
+// transient lock errors (lock-wait timeout 1205, deadlock 1213). In single-threaded
+// mode on MySQL 8.x, NOWAIT (3572) can also occur and is retried here.
 const rowCopyMaxTransientRetries = 20
 
+// rowCopyUsesNoWait reports whether row-copy should use SELECT ... FOR SHARE NOWAIT.
+// On MySQL 8.x transactional tables, NOWAIT fails fast instead of blocking behind
+// concurrent DML. Under MTS (NumWorkers > 1), lock contention with parallel DML
+// workers is expected; row-copy waits via LOCK IN SHARE MODE instead.
+func rowCopyUsesNoWait(migrationContext *base.MigrationContext) bool {
+	if !strings.HasPrefix(migrationContext.ApplierMySQLVersion, "8.") {
+		return false
+	}
+	if migrationContext.NumWorkers > 1 {
+		return false
+	}
+	return true
+}
+
 // 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) {
@@ -1131,8 +1143,7 @@ func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected i
 		apl.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
 		apl.migrationContext.GetIteration() == 0,
 		apl.migrationContext.IsTransactionalTable(),
-		// TODO: Don't hardcode this
-		strings.HasPrefix(apl.migrationContext.ApplierMySQLVersion, "8."),
+		rowCopyUsesNoWait(apl.migrationContext),
 	)
 	if err != nil {
 		return chunkSize, rowsAffected, duration, err

go/logic/applier_test.go

@@ -233,6 +233,29 @@ func TestRowCopyMaxTransientRetries(t *testing.T) {
 	require.Equal(t, 20, rowCopyMaxTransientRetries, "rowCopyMaxTransientRetries should be 20")
 }
 
+func TestRowCopyUsesNoWait(t *testing.T) {
+	t.Run("mysql8 single-threaded uses NOWAIT", func(t *testing.T) {
+		mctx := base.NewMigrationContext()
+		mctx.ApplierMySQLVersion = "8.4.3"
+		mctx.NumWorkers = 1
+		require.True(t, rowCopyUsesNoWait(mctx))
+	})
+
+	t.Run("mysql8 MTS waits for locks", func(t *testing.T) {
+		mctx := base.NewMigrationContext()
+		mctx.ApplierMySQLVersion = "8.4.3"
+		mctx.NumWorkers = 4
+		require.False(t, rowCopyUsesNoWait(mctx))
+	})
+
+	t.Run("mysql57 never uses NOWAIT", func(t *testing.T) {
+		mctx := base.NewMigrationContext()
+		mctx.ApplierMySQLVersion = "5.7.44"
+		mctx.NumWorkers = 4
+		require.False(t, rowCopyUsesNoWait(mctx))
+	})
+}
+
 func TestApplierInstantDDL(t *testing.T) {
 	migrationContext := base.NewMigrationContext()
 	migrationContext.DatabaseName = "test"