fix vitess managed fk filter logic, added tests

Signed-off-by: Harshit Gangal <harshit@planetscale.com>

go/test/endtoend/vtgate/foreignkey/fk_test.go

@@ -34,17 +34,21 @@ func TestInsertions(t *testing.T) {
 
 	// Verify that inserting data into a table that has shard scoped foreign keys works.
 	utils.Exec(t, conn, `insert into t2(id, col) values (100, 125), (1, 132)`)
+
 	// Verify that insertion fails if the data doesn't follow the fk constraint.
 	_, err := utils.ExecAllowError(t, conn, `insert into t2(id, col) values (1310, 125)`)
 	require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")
+
 	// Verify that insertion fails if the table has cross-shard foreign keys (even if the data follows the constraints).
 	_, err = utils.ExecAllowError(t, conn, `insert into t3(id, col) values (100, 100)`)
 	require.ErrorContains(t, err, "VT12002: unsupported: cross-shard foreign keys")
 
 	// insert some data in a table with multicol vindex.
 	utils.Exec(t, conn, `insert into multicol_tbl1(cola, colb, colc, msg) values (100, 'a', 'b', 'msg'), (101, 'c', 'd', 'msg2')`)
+
 	// Verify that inserting data into a table that has shard scoped multi-column foreign keys works.
 	utils.Exec(t, conn, `insert into multicol_tbl2(cola, colb, colc, msg) values (100, 'a', 'b', 'msg3')`)
+
 	// Verify that insertion fails if the data doesn't follow the fk constraint.
 	_, err = utils.ExecAllowError(t, conn, `insert into multicol_tbl2(cola, colb, colc, msg) values (103, 'c', 'd', 'msg2')`)
 	require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")

go/test/endtoend/vtgate/foreignkey/main_test.go

@@ -90,10 +90,16 @@ func start(t *testing.T) (*mysql.Conn, func()) {
 	require.NoError(t, err)
 
 	deleteAll := func() {
+		_ = utils.Exec(t, conn, "use `ks/-80`")
 		tables := []string{"t3", "t2", "t1", "multicol_tbl2", "multicol_tbl1"}
 		for _, table := range tables {
 			_ = utils.Exec(t, conn, "delete from "+table)
 		}
+		_ = utils.Exec(t, conn, "use `ks/80-`")
+		for _, table := range tables {
+			_ = utils.Exec(t, conn, "delete from "+table)
+		}
+		_ = utils.Exec(t, conn, "use `ks`")
 	}
 
 	deleteAll()

go/vt/vtgate/planbuilder/plan_test.go

@@ -435,25 +435,33 @@ func loadSchema(t testing.TB, filename string, setCollation bool) *vindexes.VSch
 	}
 	if vschema.Keyspaces["user_fk_allow"] != nil {
 		// FK from multicol_tbl2 referencing multicol_tbl1 that is shard scoped.
-		err = vschema.AddForeignKey("user_fk_allow", "multicol_tbl2", createFkDefinition([]string{"colb", "cola", "x", "colc", "y"}, "multicol_tbl1", []string{"colb", "cola", "y", "colc", "x"}))
+		err = vschema.AddForeignKey("user_fk_allow", "multicol_tbl2", createFkDefinition([]string{"colb", "cola", "x", "colc", "y"}, "multicol_tbl1", []string{"colb", "cola", "y", "colc", "x"}, sqlparser.Cascade, sqlparser.Cascade))
 		require.NoError(t, err)
 		// FK from tbl2 referencing tbl1 that is shard scoped.
-		err = vschema.AddForeignKey("user_fk_allow", "tbl2", createFkDefinition([]string{"col2"}, "tbl1", []string{"col1"}))
+		err = vschema.AddForeignKey("user_fk_allow", "tbl2", createFkDefinition([]string{"col2"}, "tbl1", []string{"col1"}, sqlparser.Restrict, sqlparser.Restrict))
 		require.NoError(t, err)
 		// FK from tbl3 referencing tbl1 that is not shard scoped.
-		err = vschema.AddForeignKey("user_fk_allow", "tbl3", createFkDefinition([]string{"coly"}, "tbl1", []string{"col1"}))
+		err = vschema.AddForeignKey("user_fk_allow", "tbl3", createFkDefinition([]string{"coly"}, "tbl1", []string{"col1"}, sqlparser.DefaultAction, sqlparser.DefaultAction))
+		require.NoError(t, err)
+		// FK from tbl4 referencing tbl5 that is shard scoped.
+		err = vschema.AddForeignKey("user_fk_allow", "tbl4", createFkDefinition([]string{"col4"}, "tbl5", []string{"col5"}, sqlparser.SetNull, sqlparser.SetNull))
+		require.NoError(t, err)
+		// FK from tbl6 referencing tbl7 that is shard scoped.
+		err = vschema.AddForeignKey("user_fk_allow", "tbl6", createFkDefinition([]string{"col6"}, "tbl7", []string{"col7"}, sqlparser.NoAction, sqlparser.NoAction))
 		require.NoError(t, err)
 	}
 	return vschema
 }
 
 // createFkDefinition is a helper function to create a Foreign key definition struct from the columns used in it provided as list of strings.
-func createFkDefinition(childCols []string, parentTableName string, parentCols []string) *sqlparser.ForeignKeyDefinition {
+func createFkDefinition(childCols []string, parentTableName string, parentCols []string, onUpdate, onDelete sqlparser.ReferenceAction) *sqlparser.ForeignKeyDefinition {
 	return &sqlparser.ForeignKeyDefinition{
 		Source: sqlparser.MakeColumns(childCols...),
 		ReferenceDefinition: &sqlparser.ReferenceDefinition{
 			ReferencedTable:   sqlparser.NewTableName(parentTableName),
 			ReferencedColumns: sqlparser.MakeColumns(parentCols...),
+			OnUpdate:          onUpdate,
+			OnDelete:          onDelete,
 		},
 	}
 }

go/vt/vtgate/planbuilder/testdata/foreignkey_cases.json

@@ -53,5 +53,42 @@
         "user_fk_allow.multicol_tbl2"
       ]
     }
+  },
+  {
+    "comment": "Delete in a table with cross-shard foreign keys disallowed",
+    "query": "delete from tbl1",
+    "plan": "VT12002: unsupported: foreign keys management at vitess"
+  },
+  {
+    "comment": "Delete in a table with shard-scoped foreign keys is allowed",
+    "query": "delete from tbl7",
+    "plan": {
+      "QueryType": "DELETE",
+      "Original": "delete from tbl7",
+      "Instructions": {
+        "OperatorType": "Delete",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user_fk_allow",
+          "Sharded": true
+        },
+        "TargetTabletType": "PRIMARY",
+        "Query": "delete from tbl7",
+        "Table": "tbl7"
+      },
+      "TablesUsed": [
+        "user_fk_allow.tbl7"
+      ]
+    }
+  },
+  {
+    "comment": "Delete in a table with shard-scoped multiple column foreign key with cascade not allowed",
+    "query": "delete from multicol_tbl1 where cola = 1 and  colb = 2 and colc = 3",
+    "plan": "VT12002: unsupported: foreign keys management at vitess"
+  },
+  {
+    "comment": "Delete in a table with shard-scoped foreign keys with cascade disallowed",
+    "query": "delete from tbl5",
+    "plan": "VT12002: unsupported: foreign keys management at vitess"
   }
 ]

go/vt/vtgate/planbuilder/testdata/vschemas/schema.json

@@ -608,8 +608,7 @@
       "foreignKeyMode": "FK_MANAGED",
       "vindexes": {
         "hash_vin": {
-          "type": "hash_test",
-          "owner": "user"
+          "type": "hash_test"
         },
         "multicolIdx": {
           "type": "multiCol_test"
@@ -663,6 +662,38 @@
               "name": "hash_vin"
             }
           ]
+        },
+        "tbl4": {
+          "column_vindexes": [
+            {
+              "column": "col4",
+              "name": "hash_vin"
+            }
+          ]
+        },
+        "tbl5": {
+          "column_vindexes": [
+            {
+              "column": "col5",
+              "name": "hash_vin"
+            }
+          ]
+        },
+        "tbl6": {
+          "column_vindexes": [
+            {
+              "column": "col6",
+              "name": "hash_vin"
+            }
+          ]
+        },
+        "tbl7": {
+          "column_vindexes": [
+            {
+              "column": "col7",
+              "name": "hash_vin"
+            }
+          ]
         }
       }
     }

go/vt/vtgate/vindexes/foreign_keys.go

@@ -136,22 +136,21 @@ func (t *Table) CrossShardParentFKs() (crossShardFKs []ParentFKInfo) {
 // This can be either the foreign key is not shard scoped or the child tables needs cascading.
 func (t *Table) ChildFKsNeedsHandling() (fks []ChildFKInfo) {
 	for _, fk := range t.ChildForeignKeys {
-		if fk.OnDelete == sqlparser.NoAction {
-			continue
-		}
 		// If the keyspaces are different, then the fk definition
 		// is going to go across shards.
 		if fk.Table.Keyspace.Name != t.Keyspace.Name {
 			fks = append(fks, fk)
 			continue
 		}
 		// If the action is not Restrict, then it needs a cascade.
-		if fk.OnDelete != sqlparser.Restrict {
+		switch fk.OnDelete {
+		case sqlparser.Cascade, sqlparser.SetNull, sqlparser.SetDefault:
 			fks = append(fks, fk)
-			continue
 		}
-
-		// Check if the fk restrict is shard scoped.
+		// sqlparser.Restrict, sqlparser.NoAction, sqlparser.DefaultAction
+		// all the actions means the same thing i.e. Restrict
+		// do not allow modification if there is a child row.
+		// Check if the restrict is shard scoped.
 		if !isShardScoped(t, fk.Table, fk.ParentColumns, fk.ChildColumns) {
 			fks = append(fks, fk)
 		}