MoveTables Atomic Mode: set FK checks off while deploying schema on t…

…argets (#15448)

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

go/test/endtoend/vreplication/fk_config_test.go

@@ -16,19 +16,28 @@ limitations under the License.
 
 package vreplication
 
+// The tables in the schema are selected so that we have one parent/child table with names in reverse lexical order
+// (child before parent), t1,t2  are in lexical order, and t11,t12  have valid circular foreign key constraints.
 var (
 	initialFKSchema = `
 create table parent(id int, name varchar(128), primary key(id)) engine=innodb;
 create table child(id int, parent_id int, name varchar(128), primary key(id), foreign key(parent_id) references parent(id) on delete cascade) engine=innodb;
 create view vparent as select * from parent;
 create table t1(id int, name varchar(128), primary key(id)) engine=innodb;
 create table t2(id int, t1id int, name varchar(128), primary key(id), foreign key(t1id) references t1(id) on delete cascade) engine=innodb;
+create table t11 (id int primary key, i int);
+create table t12 (id int primary key, i int);
+alter table t11 add constraint f11 foreign key (i) references t12 (id);
+alter table t12 add constraint f12 foreign key (i) references t11 (id);
 `
 	initialFKData = `
 insert into parent values(1, 'parent1'), (2, 'parent2');
 insert into child values(1, 1, 'child11'), (2, 1, 'child21'), (3, 2, 'child32');
 insert into t1 values(1, 't11'), (2, 't12');
 insert into t2 values(1, 1, 't21'), (2, 1, 't22'), (3, 2, 't23');
+insert into t11 values(1, null);
+insert into t12 values(1, 1);
+update t11 set i = 1 where id = 1;
 `
 
 	initialFKSourceVSchema = `
@@ -37,7 +46,9 @@ insert into t2 values(1, 1, 't21'), (2, 1, 't22'), (3, 2, 't23');
 	"parent": {},
 	"child": {},
 	"t1": {},
-	"t2": {}
+	"t2": {},
+	"t11": {},
+	"t12": {}
   }
 }
 `
@@ -82,6 +93,22 @@ insert into t2 values(1, 1, 't21'), (2, 1, 't22'), (3, 2, 't23');
           "name": "reverse_bits"
         }
       ]
+    },
+    "t11": {
+      "column_vindexes": [
+        {
+          "column": "id",
+          "name": "reverse_bits"
+        }
+      ]
+    },
+    "t12": {
+      "column_vindexes": [
+        {
+          "column": "id",
+          "name": "reverse_bits"
+        }
+      ]
     }
   }
 }

go/test/endtoend/vreplication/fk_test.go

@@ -106,7 +106,9 @@ func TestFKWorkflow(t *testing.T) {
 	require.NotNil(t, targetTab)
 	catchup(t, targetTab, workflowName, "MoveTables")
 	vdiff(t, targetKeyspace, workflowName, cellName, true, false, nil)
-	ls.waitForAdditionalRows(200)
+	if withLoad {
+		ls.waitForAdditionalRows(200)
+	}
 	vdiff(t, targetKeyspace, workflowName, cellName, true, false, nil)
 	if withLoad {
 		cancel()
@@ -121,13 +123,19 @@ func TestFKWorkflow(t *testing.T) {
 		ls = newFKLoadSimulator(t, ctx)
 		defer cancel()
 		go ls.simulateLoad()
-	}
-	ls.waitForAdditionalRows(200)
-	if withLoad {
+		ls.waitForAdditionalRows(200)
 		cancel()
 		<-ch
 	}
 	mt.Complete()
+	vtgateConn, closeConn := getVTGateConn()
+	defer closeConn()
+
+	t11Count := getRowCount(t, vtgateConn, "t11")
+	t12Count := getRowCount(t, vtgateConn, "t12")
+	require.Greater(t, t11Count, 1)
+	require.Greater(t, t12Count, 1)
+	require.Equal(t, t11Count, t12Count)
 }
 
 func insertInitialFKData(t *testing.T) {
@@ -140,11 +148,18 @@ func insertInitialFKData(t *testing.T) {
 		log.Infof("Inserting initial FK data")
 		execMultipleQueries(t, vtgateConn, db, initialFKData)
 		log.Infof("Done inserting initial FK data")
-		waitForRowCount(t, vtgateConn, db, "parent", 2)
-		waitForRowCount(t, vtgateConn, db, "child", 3)
-		waitForRowCount(t, vtgateConn, db, "t1", 2)
-		waitForRowCount(t, vtgateConn, db, "t2", 3)
 
+		type tableCounts struct {
+			name  string
+			count int
+		}
+		for _, table := range []tableCounts{
+			{"parent", 2}, {"child", 3},
+			{"t1", 2}, {"t2", 3},
+			{"t11", 1}, {"t12", 1},
+		} {
+			waitForRowCount(t, vtgateConn, db, table.name, table.count)
+		}
 	})
 }
 
@@ -170,6 +185,7 @@ func newFKLoadSimulator(t *testing.T, ctx context.Context) *fkLoadSimulator {
 	}
 }
 
+var indexCounter int = 100 // used to insert into t11 and t12
 func (ls *fkLoadSimulator) simulateLoad() {
 	t := ls.t
 	var err error
@@ -193,8 +209,13 @@ func (ls *fkLoadSimulator) simulateLoad() {
 		default: // 20% chance to delete
 			ls.delete()
 		}
+		for _, table := range []string{"t11", "t12"} {
+			query := fmt.Sprintf("insert /*+ SET_VAR(foreign_key_checks=0) */ into fksource.%s values(%d, %d)", table, indexCounter, indexCounter)
+			ls.exec(query)
+			indexCounter++
+		}
 		require.NoError(t, err)
-		time.Sleep(1 * time.Millisecond)
+		time.Sleep(10 * time.Millisecond)
 	}
 }
 

go/vt/mysqlctl/schema.go

@@ -536,6 +536,10 @@ func (mysqld *Mysqld) ApplySchemaChange(ctx context.Context, dbName string, chan
 		sql = "SET sql_log_bin = 0;\n" + sql
 	}
 
+	if change.DisableForeignKeyChecks {
+		sql = "SET foreign_key_checks = 0;\n" + sql
+	}
+
 	// add a 'use XXX' in front of the SQL
 	sql = fmt.Sprintf("USE %s;\n%s", sqlescape.EscapeID(dbName), sql)
 

go/vt/mysqlctl/tmutils/schema.go

@@ -300,12 +300,13 @@ func DiffSchemaToArray(leftName string, left *tabletmanagerdatapb.SchemaDefiniti
 // SchemaChange contains all necessary information to apply a schema change.
 // It should not be sent over the wire, it's just a set of parameters.
 type SchemaChange struct {
-	SQL              string
-	Force            bool
-	AllowReplication bool
-	BeforeSchema     *tabletmanagerdatapb.SchemaDefinition
-	AfterSchema      *tabletmanagerdatapb.SchemaDefinition
-	SQLMode          string
+	SQL                     string
+	Force                   bool
+	AllowReplication        bool
+	BeforeSchema            *tabletmanagerdatapb.SchemaDefinition
+	AfterSchema             *tabletmanagerdatapb.SchemaDefinition
+	SQLMode                 string
+	DisableForeignKeyChecks bool
 }
 
 // Equal compares two SchemaChange objects.