Add cycle detection for foreign keys

go.mod

@@ -96,6 +96,7 @@ require (
 require (
 	github.com/Shopify/toxiproxy/v2 v2.5.0
 	github.com/bndr/gotabulate v1.1.2
+	github.com/dominikbraun/graph v0.23.0
 	github.com/gammazero/deque v0.2.1
 	github.com/google/safehtml v0.1.0
 	github.com/hashicorp/go-version v1.6.0

go.sum

@@ -161,6 +161,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
+github.com/dominikbraun/graph v0.23.0 h1:TdZB4pPqCLFxYhdyMFb1TBdFxp8XLcJfTTBQucVPgCo=
+github.com/dominikbraun/graph v0.23.0/go.mod h1:yOjYyogZLY1LSG9E33JWZJiq5k83Qy2C6POAuiViluc=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=

go/test/endtoend/utils/utils.go

@@ -253,6 +253,33 @@ func WaitForAuthoritative(t *testing.T, ks, tbl string, readVSchema func() (*int
 	}
 }
 
+// WaitForKsError waits for the ks error field to be populated and returns it.
+func WaitForKsError(t *testing.T, vtgateProcess cluster.VtgateProcess, ks string) string {
+	timeout := time.After(60 * time.Second)
+	for {
+		select {
+		case <-timeout:
+			t.Fatalf("schema tracking did not find error in '%s'", ks)
+			return ""
+		default:
+			time.Sleep(1 * time.Second)
+			res, err := vtgateProcess.ReadVSchema()
+			require.NoError(t, err, res)
+			kss := convertToMap(*res)["keyspaces"]
+			ksMap := convertToMap(convertToMap(kss)[ks])
+			ksErr, fieldPresent := ksMap["error"]
+			if !fieldPresent {
+				break
+			}
+			errString, isErr := ksErr.(string)
+			if !isErr {
+				break
+			}
+			return errString
+		}
+	}
+}
+
 // WaitForColumn waits for a table's column to be present
 func WaitForColumn(t *testing.T, vtgateProcess cluster.VtgateProcess, ks, tbl, col string) error {
 	timeout := time.After(60 * time.Second)

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

@@ -774,3 +774,26 @@ func TestFkScenarios(t *testing.T) {
 		})
 	}
 }
+
+func TestCyclicFks(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+	_ = utils.Exec(t, mcmp.VtConn, "use `uks`")
+
+	// Create a cyclic foreign key constraint.
+	utils.Exec(t, mcmp.VtConn, "alter table fk_t10 add constraint test_cyclic_fks foreign key (col) references fk_t12 (col) on delete cascade on update cascade")
+	defer func() {
+		utils.Exec(t, mcmp.VtConn, "alter table fk_t10 drop foreign key test_cyclic_fks")
+	}()
+
+	// Wait for schema-tracking to be complete.
+	ksErr := utils.WaitForKsError(t, clusterInstance.VtgateProcess, unshardedKs)
+	// Make sure Vschema has the error for cyclic foreign keys.
+	assert.Contains(t, ksErr, "VT09019: uks has cyclic foreign keys")
+
+	// Ensure that the Vitess database is originally empty
+	ensureDatabaseState(t, mcmp.VtConn, true)
+
+	_, err := utils.ExecAllowError(t, mcmp.VtConn, "insert into fk_t10(id, col) values (1, 1)")
+	require.ErrorContains(t, err, "VT09019: uks has cyclic foreign keys")
+}

go/test/vschemawrapper/vschema_wrapper.go

@@ -136,6 +136,10 @@ func (vw *VSchemaWrapper) ForeignKeyMode(keyspace string) (vschemapb.Keyspace_Fo
 	return defaultFkMode, nil
 }
 
+func (vw *VSchemaWrapper) KeyspaceError(keyspace string) error {
+	return nil
+}
+
 func (vw *VSchemaWrapper) AllKeyspace() ([]*vindexes.Keyspace, error) {
 	if vw.Keyspace == nil {
 		return nil, vterrors.VT13001("keyspace not available")

go/vt/schemadiff/semantics.go

@@ -60,6 +60,10 @@ func (si *declarativeSchemaInformation) ForeignKeyMode(keyspace string) (vschema
 	return vschemapb.Keyspace_unmanaged, nil
 }
 
+func (si *declarativeSchemaInformation) KeyspaceError(keyspace string) error {
+	return nil
+}
+
 // addTable adds a fake table with an empty column list
 func (si *declarativeSchemaInformation) addTable(tableName string) {
 	tbl := &vindexes.Table{

go/vt/vterrors/code.go

@@ -79,7 +79,8 @@ var (
 	VT09015 = errorWithoutState("VT09015", vtrpcpb.Code_FAILED_PRECONDITION, "schema tracking required", "This query cannot be planned without more information on the SQL schema. Please turn on schema tracking or add authoritative columns information to your VSchema.")
 	VT09016 = errorWithState("VT09016", vtrpcpb.Code_FAILED_PRECONDITION, RowIsReferenced2, "Cannot delete or update a parent row: a foreign key constraint fails", "SET DEFAULT is not supported by InnoDB")
 	VT09017 = errorWithoutState("VT09017", vtrpcpb.Code_FAILED_PRECONDITION, "%s", "Invalid syntax for the statement type.")
-	VT09018 = errorWithoutState("VT09017", vtrpcpb.Code_FAILED_PRECONDITION, "%s", "Invalid syntax for the vindex function statement.")
+	VT09018 = errorWithoutState("VT09018", vtrpcpb.Code_FAILED_PRECONDITION, "%s", "Invalid syntax for the vindex function statement.")
+	VT09019 = errorWithoutState("VT09019", vtrpcpb.Code_FAILED_PRECONDITION, "%s has cyclic foreign keys", "Vitess doesn't support cyclic foreign keys.")
 
 	VT10001 = errorWithoutState("VT10001", vtrpcpb.Code_ABORTED, "foreign key constraints are not allowed", "Foreign key constraints are not allowed, see https://vitess.io/blog/2021-06-15-online-ddl-why-no-fk/.")
 

go/vt/vtgate/planbuilder/plancontext/vschema.go

@@ -57,6 +57,9 @@ type VSchema interface {
 	// ForeignKeyMode returns the foreign_key flag value
 	ForeignKeyMode(keyspace string) (vschemapb.Keyspace_ForeignKeyMode, error)
 
+	// KeyspaceError returns any error in the keyspace vschema.
+	KeyspaceError(keyspace string) error
+
 	// GetVSchema returns the latest cached vindexes.VSchema
 	GetVSchema() *vindexes.VSchema
 

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

@@ -426,46 +426,46 @@
   {
     "comment": "select keyspace_id from user_index where id = 1 and id = 2",
     "query": "select keyspace_id from user_index where id = 1 and id = 2",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (multiple filters)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (multiple filters)"
   },
   {
     "comment": "select keyspace_id from user_index where func(id)",
     "query": "select keyspace_id from user_index where func(id)",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (not a comparison)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (not a comparison)"
   },
   {
     "comment": "select keyspace_id from user_index where id > 1",
     "query": "select keyspace_id from user_index where id > 1",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (not equality)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (not equality)"
   },
   {
     "comment": "select keyspace_id from user_index where 1 = id",
     "query": "select keyspace_id from user_index where 1 = id",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (lhs is not a column)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (lhs is not a column)"
   },
   {
     "comment": "select keyspace_id from user_index where keyspace_id = 1",
     "query": "select keyspace_id from user_index where keyspace_id = 1",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (lhs is not id)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (lhs is not id)"
   },
   {
     "comment": "select keyspace_id from user_index where id = id+1",
     "query": "select keyspace_id from user_index where id = id+1",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (rhs is not a value)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (rhs is not a value)"
   },
   {
     "comment": "vindex func without where condition",
     "query": "select keyspace_id from user_index",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (where clause missing)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (where clause missing)"
   },
   {
     "comment": "vindex func in subquery without where",
     "query": "select id from user where exists(select keyspace_id from user_index)",
-    "plan": "VT09017: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (where clause missing)"
+    "plan": "VT09018: WHERE clause for vindex function must be of the form id = <val> or id in(<val>,...) (where clause missing)"
   },
   {
     "comment": "select func(keyspace_id) from user_index where id = :id",
     "query": "select func(keyspace_id) from user_index where id = :id",
-    "plan": "VT09017: cannot add 'func(keyspace_id)' expression to a table/vindex"
+    "plan": "VT09018: cannot add 'func(keyspace_id)' expression to a table/vindex"
   }
 ]

go/vt/vtgate/semantics/FakeSI.go

@@ -34,6 +34,7 @@ type FakeSI struct {
 	Tables           map[string]*vindexes.Table
 	VindexTables     map[string]vindexes.Vindex
 	KsForeignKeyMode map[string]vschemapb.Keyspace_ForeignKeyMode
+	KsError          map[string]error
 }
 
 // FindTableOrVindex implements the SchemaInformation interface
@@ -59,3 +60,14 @@ func (s *FakeSI) ForeignKeyMode(keyspace string) (vschemapb.Keyspace_ForeignKeyM
 	}
 	return vschemapb.Keyspace_unmanaged, nil
 }
+
+func (s *FakeSI) KeyspaceError(keyspace string) error {
+	if s.KsError != nil {
+		fkErr, isPresent := s.KsError[keyspace]
+		if !isPresent {
+			return fmt.Errorf("%v keyspace not found", keyspace)
+		}
+		return fkErr
+	}
+	return nil
+}

go/vt/vtgate/semantics/analyzer.go

@@ -469,6 +469,11 @@ func (a *analyzer) getAllManagedForeignKeys() (map[TableSet][]vindexes.ChildFKIn
 		if fkMode != vschemapb.Keyspace_managed {
 			continue
 		}
+		// Cyclic foreign key constraints error is stored in the keyspace.
+		ksErr := a.tables.si.KeyspaceError(vi.Keyspace.Name)
+		if ksErr != nil {
+			return nil, nil, ksErr
+		}
 
 		// Add all the child and parent foreign keys to our map.
 		ts := SingleTableSet(idx)

go/vt/vtgate/semantics/analyzer_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package semantics
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -1610,6 +1611,27 @@ func TestGetAllManagedForeignKeys(t *testing.T) {
 			},
 			expectedErr: "undefined_ks keyspace not found",
 		},
+		{
+			name: "Cyclic fk constraints error",
+			analyzer: &analyzer{
+				tables: &tableCollector{
+					Tables: []TableInfo{
+						tbl["t0"], tbl["t1"],
+						&DerivedTable{},
+					},
+					si: &FakeSI{
+						KsForeignKeyMode: map[string]vschemapb.Keyspace_ForeignKeyMode{
+							"ks":           vschemapb.Keyspace_managed,
+							"ks_unmanaged": vschemapb.Keyspace_unmanaged,
+						},
+						KsError: map[string]error{
+							"ks": fmt.Errorf("VT09019: ks has cyclic foreign keys"),
+						},
+					},
+				},
+			},
+			expectedErr: "VT09019: ks has cyclic foreign keys",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

go/vt/vtgate/semantics/info_schema.go

@@ -1717,3 +1717,7 @@ func (i *infoSchemaWithColumns) ConnCollation() collations.ID {
 func (i *infoSchemaWithColumns) ForeignKeyMode(keyspace string) (vschemapb.Keyspace_ForeignKeyMode, error) {
 	return i.inner.ForeignKeyMode(keyspace)
 }
+
+func (i *infoSchemaWithColumns) KeyspaceError(keyspace string) error {
+	return i.inner.KeyspaceError(keyspace)
+}

go/vt/vtgate/semantics/semantic_state.go

@@ -145,6 +145,7 @@ type (
 		ConnCollation() collations.ID
 		// ForeignKeyMode returns the foreign_key flag value
 		ForeignKeyMode(keyspace string) (vschemapb.Keyspace_ForeignKeyMode, error)
+		KeyspaceError(keyspace string) error
 	}
 )
 

go/vt/vtgate/vcursor_impl.go

@@ -1047,6 +1047,14 @@ func (vc *vcursorImpl) ForeignKeyMode(keyspace string) (vschemapb.Keyspace_Forei
 	return ks.ForeignKeyMode, nil
 }
 
+func (vc *vcursorImpl) KeyspaceError(keyspace string) error {
+	ks := vc.vschema.Keyspaces[keyspace]
+	if ks == nil {
+		return vterrors.VT14004(keyspace)
+	}
+	return ks.Error
+}
+
 // ParseDestinationTarget parses destination target string and sets default keyspace if possible.
 func parseDestinationTarget(targetString string, vschema *vindexes.VSchema) (string, topodatapb.TabletType, key.Destination, error) {
 	destKeyspace, destTabletType, dest, err := topoprotopb.ParseDestination(targetString, defaultTabletType)

go/vt/vtgate/vschema_manager.go

@@ -20,11 +20,14 @@ import (
 	"context"
 	"sync"
 
+	"github.com/dominikbraun/graph"
+
 	"vitess.io/vitess/go/vt/log"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/srvtopo"
 	"vitess.io/vitess/go/vt/topo"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 
 	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
@@ -188,6 +191,9 @@ func (vm *VSchemaManager) buildAndEnhanceVSchema(v *vschemapb.SrvVSchema) *vinde
 	vschema := vindexes.BuildVSchema(v)
 	if vm.schema != nil {
 		vm.updateFromSchema(vschema)
+		// We mark the keyspaces that have foreign key management in Vitess and have cyclic foreign keys
+		// to have an error. This makes all queries against them to fail.
+		markErrorIfCyclesInFk(vschema)
 	}
 	return vschema
 }
@@ -231,6 +237,49 @@ func (vm *VSchemaManager) updateFromSchema(vschema *vindexes.VSchema) {
 	}
 }
 
+type tableCol struct {
+	tableName string
+	colNames  sqlparser.Columns
+}
+
+var tableColHash = func(tc tableCol) string {
+	res := tc.tableName
+	for _, colName := range tc.colNames {
+		res += "|" + colName.Lowered()
+	}
+	return res
+}
+
+func markErrorIfCyclesInFk(vschema *vindexes.VSchema) {
+	for ksName, ks := range vschema.Keyspaces {
+		// Only check cyclic foreign keys for keyspaces that have
+		// foreign keys managed in Vitess.
+		if ks.ForeignKeyMode != vschemapb.Keyspace_managed {
+			continue
+		}
+		g := graph.New(tableColHash, graph.PreventCycles(), graph.Directed())
+		for tableName, table := range ks.Tables {
+			for _, cfk := range table.ChildForeignKeys {
+				childTable := cfk.Table
+				parentVertex := tableCol{
+					tableName: tableName,
+					colNames:  cfk.ParentColumns,
+				}
+				childVertex := tableCol{
+					tableName: childTable.Name.String(),
+					colNames:  cfk.ChildColumns,
+				}
+				_ = g.AddVertex(parentVertex)
+				_ = g.AddVertex(childVertex)
+				err := g.AddEdge(tableColHash(parentVertex), tableColHash(childVertex))
+				if err != nil {
+					ks.Error = vterrors.VT09019(ksName)
+				}
+			}
+		}
+	}
+}
+
 func setColumns(ks *vindexes.KeyspaceSchema, tblName string, columns []vindexes.Column) *vindexes.Table {
 	vTbl := ks.Tables[tblName]
 	if vTbl == nil {