[release-19.0] VReplication Atomic Copy Workflows: fix bugs around concurrent inserts (#17772)

go/test/endtoend/vreplication/fk_ext_test.go

@@ -90,7 +90,10 @@ func TestFKExt(t *testing.T) {
 	setSidecarDBName("_vt")
 
 	// Ensure that there are multiple copy phase cycles per table.
-	extraVTTabletArgs = append(extraVTTabletArgs, "--vstream_packet_size=256", "--queryserver-config-schema-change-signal")
+	extraVTTabletArgs = append(extraVTTabletArgs,
+		"--vstream_packet_size=256",
+		"--queryserver-config-schema-change-signal",
+		parallelInsertWorkers)
 	extraVTGateArgs = append(extraVTGateArgs, "--schema_change_signal=true", "--planner-version", "Gen4")
 	defer func() { extraVTTabletArgs = nil }()
 	initFKExtConfig(t)

go/test/endtoend/vreplication/fk_test.go

@@ -40,6 +40,7 @@ const testWorkflowFlavor = workflowFlavorRandom
 // It inserts initial data, then simulates load. We insert both child rows with foreign keys and those without,
 // i.e. with foreign_key_checks=0.
 func TestFKWorkflow(t *testing.T) {
+	setSidecarDBName("_vt")
 	extraVTTabletArgs = []string{
 		// Ensure that there are multiple copy phase cycles per table.
 		"--vstream_packet_size=256",

go/vt/vttablet/tabletmanager/vreplication/vcopier_atomic.go

@@ -86,11 +86,7 @@
 	defer rowsCopiedTicker.Stop()
 
 	parallelism := getInsertParallelism()
-	// For now do not support concurrent inserts for atomic copies.
-	if parallelism > 1 {
-		parallelism = 1
-		log.Infof("Disabling concurrent inserts for atomic copies")
-	}
+
 	copyWorkerFactory := vc.newCopyWorkerFactory(parallelism)
 	var copyWorkQueue *vcopierCopyWorkQueue
 
@@ -114,7 +110,6 @@
 			resp.TableName, len(resp.Fields), len(resp.Rows), resp.Gtid, resp.Lastpk)
 		tableName := resp.TableName
 		gtid = resp.Gtid
-
 		updateRowsCopied := func() error {
 			updateRowsQuery := binlogplayer.GenerateUpdateRowsCopied(vc.vr.id, vc.vr.stats.CopyRowCount.Get())
 			_, err := vc.vr.dbClient.Execute(updateRowsQuery)
@@ -204,6 +199,10 @@
 		log.Infof("copying table %s with lastpk %v", tableName, lastpkbv)
 		// Prepare a vcopierCopyTask for the current batch of work.
 		currCh := make(chan *vcopierCopyTaskResult, 1)
+
+		if parallelism > 1 {
+			resp = resp.CloneVT()
+		}
 		currT := newVCopierCopyTask(newVCopierCopyTaskArgs(resp.Rows, resp.Lastpk))
 
 		// Send result to the global resultCh and currCh. resultCh is used by
@@ -291,12 +290,12 @@
 		log.Infof("Copy of %v stopped", state.currentTableName)
 		return fmt.Errorf("CopyAll was interrupted due to context expiration")
 	default:
-		if err := vc.deleteCopyState(state.currentTableName); err != nil {
-			return err
-		}
 		if copyWorkQueue != nil {
 			copyWorkQueue.close()
 		}
+		if err := vc.deleteCopyState(state.currentTableName); err != nil {
+			return err
+		}
 		if err := vc.updatePos(ctx, gtid); err != nil {
 			return err
 		}