schemadiff: validate and apply foreign key indexes

go/vt/schemadiff/errors.go

@@ -301,6 +301,32 @@ func (e *ForeignKeyColumnTypeMismatchError) Error() string {
 	)
 }
 
+type MissingForeignKeyReferencedIndexError struct {
+	Table           string
+	Constraint      string
+	ReferencedTable string
+}
+
+func (e *MissingForeignKeyReferencedIndexError) Error() string {
+	return fmt.Sprintf("missing index in referenced table %s for foreign key constraint %s in table %s",
+		sqlescape.EscapeID(e.ReferencedTable),
+		sqlescape.EscapeID(e.Constraint),
+		sqlescape.EscapeID(e.Table),
+	)
+}
+
+type IndexNeededByForeignKeyError struct {
+	Table string
+	Key   string
+}
+
+func (e *IndexNeededByForeignKeyError) Error() string {
+	return fmt.Sprintf("key %s needed by a foreign key constraint in table %s",
+		sqlescape.EscapeID(e.Key),
+		sqlescape.EscapeID(e.Table),
+	)
+}
+
 type ViewDependencyUnresolvedError struct {
 	View string
 }

go/vt/schemadiff/schema.go

@@ -368,7 +368,9 @@ func (s *Schema) normalize() error {
 				}
 			}
 
-			// TODO(shlomi): find a valid index
+			if !referencedTable.columnsCoveredByInOrderIndex(check.ReferenceDefinition.ReferencedColumns) {
+				return &MissingForeignKeyReferencedIndexError{Table: t.Name(), Constraint: cs.Name.String(), ReferencedTable: referencedTableName}
+			}
 		}
 	}
 	return nil

go/vt/schemadiff/schema_test.go

@@ -249,15 +249,15 @@ func TestGetForeignKeyParentTableNames(t *testing.T) {
 
 func TestTableForeignKeyOrdering(t *testing.T) {
 	fkQueries := []string{
-		"create table t11 (id int primary key, i int, constraint f12 foreign key (i) references t12(id) on delete restrict, constraint f20 foreign key (i) references t20(id) on delete restrict)",
+		"create table t11 (id int primary key, i int, key ix (i), constraint f12 foreign key (i) references t12(id) on delete restrict, constraint f20 foreign key (i) references t20(id) on delete restrict)",
 		"create table t15(id int, primary key(id))",
 		"create view v09 as select * from v13, t17",
-		"create table t20 (id int primary key, i int, constraint f15 foreign key (i) references t15(id) on delete restrict)",
+		"create table t20 (id int primary key, i int, key ix (i), constraint f15 foreign key (i) references t15(id) on delete restrict)",
 		"create view v13 as select * from t20",
-		"create table t12 (id int primary key, i int, constraint f15 foreign key (i) references t15(id) on delete restrict)",
-		"create table t17 (id int primary key, i int, constraint f11 foreign key (i) references t11(id) on delete restrict, constraint f15 foreign key (i) references t15(id) on delete restrict)",
-		"create table t16 (id int primary key, i int, constraint f11 foreign key (i) references t11(id) on delete restrict, constraint f15 foreign key (i) references t15(id) on delete restrict)",
-		"create table t14 (id int primary key, i int, constraint f14 foreign key (i) references t14(id) on delete restrict)",
+		"create table t12 (id int primary key, i int, key ix (i), constraint f15 foreign key (i) references t15(id) on delete restrict)",
+		"create table t17 (id int primary key, i int, key ix (i), constraint f11 foreign key (i) references t11(id) on delete restrict, constraint f15 foreign key (i) references t15(id) on delete restrict)",
+		"create table t16 (id int primary key, i int, key ix (i), constraint f11 foreign key (i) references t11(id) on delete restrict, constraint f15 foreign key (i) references t15(id) on delete restrict)",
+		"create table t14 (id int primary key, i int, key ix (i), constraint f14 foreign key (i) references t14(id) on delete restrict)",
 	}
 	expectSortedTableNames := []string{
 		"t14",
@@ -287,10 +287,10 @@ func TestInvalidSchema(t *testing.T) {
 		expectErr error
 	}{
 		{
-			schema: "create table t11 (id int primary key, i int, constraint f11 foreign key (i) references t11(id) on delete restrict)",
+			schema: "create table t11 (id int primary key, i int, key ix(i), constraint f11 foreign key (i) references t11(id) on delete restrict)",
 		},
 		{
-			schema: "create table t10(id int primary key); create table t11 (id int primary key, i int, constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema: "create table t10(id int primary key); create table t11 (id int primary key, i int, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 		},
 		{
 			schema:    "create table t11 (id int primary key, i int, constraint f11 foreign key (i7) references t11(id) on delete restrict)",
@@ -309,35 +309,39 @@ func TestInvalidSchema(t *testing.T) {
 			expectErr: &ForeignKeyDependencyUnresolvedError{Table: "t11"},
 		},
 		{
-			schema:    "create table t11 (id int primary key, i int, constraint f11 foreign key (i) references t11(id2) on delete restrict)",
+			schema:    "create table t11 (id int primary key, i int, key ix(i), constraint f11 foreign key (i) references t11(id2) on delete restrict)",
 			expectErr: &InvalidReferencedColumnInForeignKeyConstraintError{Table: "t11", Constraint: "f11", ReferencedTable: "t11", ReferencedColumn: "id2"},
 		},
 		{
-			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i int, constraint f10 foreign key (i) references t10(x) on delete restrict)",
+			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i int, key ix(i), constraint f10 foreign key (i) references t10(x) on delete restrict)",
 			expectErr: &InvalidReferencedColumnInForeignKeyConstraintError{Table: "t11", Constraint: "f10", ReferencedTable: "t10", ReferencedColumn: "x"},
 		},
 		{
-			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i int unsigned, constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema:    "create table t10(id int primary key, i int); create table t11 (id int primary key, i int, key ix(i), constraint f10 foreign key (i) references t10(i) on delete restrict)",
+			expectErr: &MissingForeignKeyReferencedIndexError{Table: "t11", Constraint: "f10", ReferencedTable: "t10"},
+		},
+		{
+			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i int unsigned, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 			expectErr: &ForeignKeyColumnTypeMismatchError{Table: "t11", Constraint: "f10", Column: "i", ReferencedTable: "t10", ReferencedColumn: "id"},
 		},
 		{
-			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i bigint, constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema:    "create table t10(id int primary key); create table t11 (id int primary key, i bigint, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 			expectErr: &ForeignKeyColumnTypeMismatchError{Table: "t11", Constraint: "f10", Column: "i", ReferencedTable: "t10", ReferencedColumn: "id"},
 		},
 		{
-			schema:    "create table t10(id bigint primary key); create table t11 (id int primary key, i int, constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema:    "create table t10(id bigint primary key); create table t11 (id int primary key, i int, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 			expectErr: &ForeignKeyColumnTypeMismatchError{Table: "t11", Constraint: "f10", Column: "i", ReferencedTable: "t10", ReferencedColumn: "id"},
 		},
 		{
-			schema:    "create table t10(id bigint primary key); create table t11 (id int primary key, i varchar(100), constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema:    "create table t10(id bigint primary key); create table t11 (id int primary key, i varchar(100), key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 			expectErr: &ForeignKeyColumnTypeMismatchError{Table: "t11", Constraint: "f10", Column: "i", ReferencedTable: "t10", ReferencedColumn: "id"},
 		},
 		{
 			// InnoDB allows different string length
-			schema: "create table t10(id varchar(50) primary key); create table t11 (id int primary key, i varchar(100), constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema: "create table t10(id varchar(50) primary key); create table t11 (id int primary key, i varchar(100), key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 		},
 		{
-			schema:    "create table t10(id varchar(50) charset utf8mb3 primary key); create table t11 (id int primary key, i varchar(100) charset utf8mb4, constraint f10 foreign key (i) references t10(id) on delete restrict)",
+			schema:    "create table t10(id varchar(50) charset utf8mb3 primary key); create table t11 (id int primary key, i varchar(100) charset utf8mb4, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
 			expectErr: &ForeignKeyColumnTypeMismatchError{Table: "t11", Constraint: "f10", Column: "i", ReferencedTable: "t10", ReferencedColumn: "id"},
 		},
 	}

go/vt/schemadiff/table.go

@@ -296,7 +296,9 @@ func NewCreateTableEntity(c *sqlparser.CreateTable) (*CreateTableEntity, error)
 // - table option case (upper/lower/special)
 // The function returns this receiver as courtesy
 func (c *CreateTableEntity) normalize() *CreateTableEntity {
-	c.normalizeKeys()
+	c.normalizePrimaryKeyColumns()
+	c.normalizeForeignKeyIndexes() // implicitly add missing indexes for foreign keys
+	c.normalizeKeys()              // assign names to keys
 	c.normalizeUnnamedConstraints()
 	c.normalizeTableOptions()
 	c.normalizeColumnOptions()
@@ -622,7 +624,7 @@ func (c *CreateTableEntity) normalizeKeys() {
 }
 
 func (c *CreateTableEntity) normalizeUnnamedConstraints() {
-	// let's ensure all keys have names
+	// let's ensure all constraints have names
 	constraintNameExists := map[string]bool{}
 	// first, we iterate and take note for all keys that do already have names
 	for _, constraint := range c.CreateTable.TableSpec.Constraints {
@@ -650,6 +652,34 @@ func (c *CreateTableEntity) normalizeUnnamedConstraints() {
 	}
 }
 
+func (c *CreateTableEntity) normalizeForeignKeyIndexes() {
+	for _, constraint := range c.CreateTable.TableSpec.Constraints {
+		fk, ok := constraint.Details.(*sqlparser.ForeignKeyDefinition)
+		if !ok {
+			continue
+		}
+		if !c.columnsCoveredByInOrderIndex(fk.Source) {
+			// We add a foreign key, but the local FK columns are not indexed.
+			// MySQL's behavior is to implicitly add an index that covers the foreign key's local columns.
+			// The name of the index is either:
+			// - the same name of the constraint, if such name is provided
+			//   - and error if an index by this name exists
+			// - or, a standard auto-generated index name, if the constraint name is not provided
+			indexDefinition := &sqlparser.IndexDefinition{
+				Info: &sqlparser.IndexInfo{
+					Type: "key",
+					Name: constraint.Name, // if name is empty, then the name is later auto populated
+				},
+			}
+			for _, col := range fk.Source {
+				indexColumn := &sqlparser.IndexColumn{Column: col}
+				indexDefinition.Columns = append(indexDefinition.Columns, indexColumn)
+			}
+			c.TableSpec.Indexes = append(c.TableSpec.Indexes, indexDefinition)
+		}
+	}
+}
+
 // Name implements Entity interface
 func (c *CreateTableEntity) Name() string {
 	return c.CreateTable.GetTable().Name.String()
@@ -1734,6 +1764,21 @@ func (c *CreateTableEntity) apply(diff *AlterTableEntityDiff) error {
 			if !found {
 				return &ApplyKeyNotFoundError{Table: c.Name(), Key: opt.Name.String()}
 			}
+
+			// Now, if this is a normal key being dropped, let's validate it does not leave any foreign key constraint uncovered
+			switch opt.Type {
+			case sqlparser.PrimaryKeyType, sqlparser.NormalKeyType:
+				for _, cs := range c.CreateTable.TableSpec.Constraints {
+					fk, ok := cs.Details.(*sqlparser.ForeignKeyDefinition)
+					if !ok {
+						continue
+					}
+					if !c.columnsCoveredByInOrderIndex(fk.Source) {
+						return &IndexNeededByForeignKeyError{Table: c.Name(), Key: opt.Name.String()}
+					}
+				}
+			}
+
 		case *sqlparser.AddIndexDefinition:
 			// validate no existing key by same name
 			keyName := opt.IndexDefinition.Info.Name.String()
@@ -1966,6 +2011,7 @@ func (c *CreateTableEntity) Apply(diff EntityDiff) (Entity, error) {
 //   - edit or remove keys if referenced columns are dropped
 //   - drop check constraints for a single specific column if that column
 //     is the only referenced column in that check constraint.
+//   - add implicit keys for foreign key constraint, if needed
 func (c *CreateTableEntity) postApplyNormalize() error {
 	// reduce or remove keys based on existing column list
 	// (a column may have been removed)postApplyNormalize
@@ -2027,6 +2073,10 @@ func (c *CreateTableEntity) postApplyNormalize() error {
 	}
 	c.CreateTable.TableSpec.Constraints = keptConstraints
 
+	c.normalizePrimaryKeyColumns()
+	c.normalizeForeignKeyIndexes()
+	c.normalizeKeys()
+
 	return nil
 }
 
@@ -2057,6 +2107,44 @@ func getKeyColumnNames(key *sqlparser.IndexDefinition) (colNames map[string]bool
 	return colNames
 }
 
+// indexCoversColumnsInOrder checks if the given index covers the given columns in order and in prefix.
+// the index must either covers the exact list of columns or continue to cover additional columns beyond.
+// Used for validating indexes covering foreign keys.
+func indexCoversColumnsInOrder(index *sqlparser.IndexDefinition, columns sqlparser.Columns) bool {
+	if len(columns) == 0 {
+		return false
+	}
+	if len(index.Columns) < len(columns) {
+		// obviously the index doesn't cover the required columns
+		return false
+	}
+	for i, col := range columns {
+		// the index must cover same columns, in order, wih possibly more columns covered than requested.
+		indexCol := index.Columns[i]
+		if !strings.EqualFold(col.String(), indexCol.Column.String()) {
+			return false
+		}
+	}
+	return true
+}
+
+// indexesCoveringForeignKeyColumns returns a list of indexes that cover a given list of coumns, in-oder and in prefix.
+// Used for validating indexes covering foreign keys.
+func (c *CreateTableEntity) indexesCoveringForeignKeyColumns(columns sqlparser.Columns) (indexes []*sqlparser.IndexDefinition) {
+	for _, index := range c.CreateTable.TableSpec.Indexes {
+		if indexCoversColumnsInOrder(index, columns) {
+			indexes = append(indexes, index)
+		}
+	}
+	return indexes
+}
+
+// columnsCoveredByInOrderIndex returns 'true' when there is at least one index that covers the given
+// list of columns in-order and in-prefix.
+func (c *CreateTableEntity) columnsCoveredByInOrderIndex(columns sqlparser.Columns) bool {
+	return len(c.indexesCoveringForeignKeyColumns(columns)) > 0
+}
+
 func (c *CreateTableEntity) validateDuplicateKeyNameError() error {
 	keyNames := map[string]bool{}
 	for _, key := range c.CreateTable.TableSpec.Indexes {
@@ -2080,6 +2168,22 @@ func (c *CreateTableEntity) validate() error {
 		}
 		columnExists[colName] = true
 	}
+	// validate all columns used by foreign key constraints do in fact exist,
+	// and that there exists an index over those columns
+	for _, cs := range c.CreateTable.TableSpec.Constraints {
+		fk, ok := cs.Details.(*sqlparser.ForeignKeyDefinition)
+		if !ok {
+			continue
+		}
+		if len(fk.Source) != len(fk.ReferenceDefinition.ReferencedColumns) {
+			return &ForeignKeyColumnCountMismatchError{Table: c.Name(), Constraint: cs.Name.String(), ColumnCount: len(fk.Source), ReferencedTable: fk.ReferenceDefinition.ReferencedTable.Name.String(), ReferencedColumnCount: len(fk.ReferenceDefinition.ReferencedColumns)}
+		}
+		for _, col := range fk.Source {
+			if !columnExists[col.Lowered()] {
+				return &InvalidColumnInForeignKeyConstraintError{Table: c.Name(), Constraint: cs.Name.String(), Column: col.String()}
+			}
+		}
+	}
 	// validate all columns referenced by indexes do in fact exist
 	for _, key := range c.CreateTable.TableSpec.Indexes {
 		for colName := range getKeyColumnNames(key) {
@@ -2130,23 +2234,6 @@ func (c *CreateTableEntity) validate() error {
 			}
 		}
 	}
-	// validate all columns used by foreign key constraints do in fact exist,
-	// and that there exists an index over those columns
-	for _, cs := range c.CreateTable.TableSpec.Constraints {
-		check, ok := cs.Details.(*sqlparser.ForeignKeyDefinition)
-		if !ok {
-			continue
-		}
-		if len(check.Source) != len(check.ReferenceDefinition.ReferencedColumns) {
-			return &ForeignKeyColumnCountMismatchError{Table: c.Name(), Constraint: cs.Name.String(), ColumnCount: len(check.Source), ReferencedTable: check.ReferenceDefinition.ReferencedTable.Name.String(), ReferencedColumnCount: len(check.ReferenceDefinition.ReferencedColumns)}
-		}
-		for _, col := range check.Source {
-			if !columnExists[col.Lowered()] {
-				return &InvalidColumnInForeignKeyConstraintError{Table: c.Name(), Constraint: cs.Name.String(), Column: col.String()}
-			}
-		}
-		// TODO(shlomi): find a valid index
-	}
 	// validate all columns referenced by constraint checks do in fact exist
 	for _, cs := range c.CreateTable.TableSpec.Constraints {
 		check, ok := cs.Details.(*sqlparser.CheckConstraintDefinition)