schemadiff: fix index/foreign-key apply scenario

go/test/endtoend/onlineddl/vrepl_suite/testdata/swap-uk-uk/alter

@@ -1 +1 @@
-drop key id_uidx, drop key its_uidx, add unique key its2_uidx(i, ts), add unique key id2_uidx(id)
+drop key id_uidx, drop key its_uidx, add unique key id2_uidx(id), add unique key its2_uidx(i, ts)

go/vt/schemadiff/diff_test.go

@@ -911,6 +911,17 @@ func TestDiffSchemas(t *testing.T) {
 				"DROP TABLE `t7`",
 			},
 		},
+		{
+			name: "rename index used by foreign keys",
+			from: `create table parent (id int primary key); create table t (id int primary key, i int, key i_idx (i), constraint f foreign key (i) references parent(id))`,
+			to:   `create table parent (id int primary key); create table t (id int primary key, i int, key i_alternative (i), constraint f foreign key (i) references parent(id))`,
+			diffs: []string{
+				"alter table t rename index i_idx to i_alternative",
+			},
+			cdiffs: []string{
+				"ALTER TABLE `t` RENAME INDEX `i_idx` TO `i_alternative`",
+			},
+		},
 		// Views
 		{
 			name: "identical views",

go/vt/schemadiff/table.go

@@ -1590,6 +1590,21 @@ func (c *CreateTableEntity) diffKeys(alterTable *sqlparser.AlterTable,
 		t2KeysMap[key.Info.Name.String()] = key
 	}
 
+	// A map of index definition text to lost of key names that have that definition.
+	// For example, in:
+	//
+	// create table t (
+	// 	i1 int,
+	// 	i2 int,
+	// 	key k1 (i1),
+	// 	key k2 (i2),
+	// 	key k3 (i2)
+	// )
+	// We will have:
+	// - "KEY `` (i1)": ["k1"]
+	// - "KEY `` (i2)": ["k2", "k3"]
+	droppedKeysAnonymousDefinitions := map[string]([]string){}
+
 	dropKeyStatement := func(info *sqlparser.IndexInfo) *sqlparser.DropKey {
 		dropKey := &sqlparser.DropKey{}
 		if info.Type == sqlparser.IndexTypePrimary {
@@ -1601,14 +1616,25 @@ func (c *CreateTableEntity) diffKeys(alterTable *sqlparser.AlterTable,
 		return dropKey
 	}
 
+	anonymizedIndexDefinition := func(indexDefinition *sqlparser.IndexDefinition) string {
+		currentName := indexDefinition.Info.Name
+		defer func() { indexDefinition.Info.Name = currentName }()
+		indexDefinition.Info.Name = sqlparser.NewIdentifierCI("")
+		return sqlparser.CanonicalString(indexDefinition)
+	}
+
 	// evaluate dropped keys
 	//
+	dropKeyStatements := map[string]*sqlparser.DropKey{}
 	for _, t1Key := range t1Keys {
 		if _, ok := t2KeysMap[t1Key.Info.Name.String()]; !ok {
 			// column exists in t1 but not in t2, hence it is dropped
 			dropKey := dropKeyStatement(t1Key.Info)
-			alterTable.AlterOptions = append(alterTable.AlterOptions, dropKey)
+			dropKeyStatements[t1Key.Info.Name.String()] = dropKey
 			annotations.MarkRemoved(sqlparser.CanonicalString(t1Key))
+
+			anonymized := anonymizedIndexDefinition(t1Key)
+			droppedKeysAnonymousDefinitions[anonymized] = append(droppedKeysAnonymousDefinitions[anonymized], t1Key.Info.Name.String())
 		}
 	}
 
@@ -1644,6 +1670,28 @@ func (c *CreateTableEntity) diffKeys(alterTable *sqlparser.AlterTable,
 			}
 		} else {
 			// key exists in t2 but not in t1, hence it is added
+
+			// But wait! As an optimization, if this index has the exact same definition as a previously dropped index,
+			// then we convert the drop+add statements in to a `RENAME INDEX` statement.
+			convertedToRename := false
+			anonymized := anonymizedIndexDefinition(t2Key)
+			if droppedKeys := droppedKeysAnonymousDefinitions[anonymized]; len(droppedKeys) > 0 {
+				if dropKey, ok := dropKeyStatements[droppedKeys[0]]; ok {
+					delete(dropKeyStatements, droppedKeys[0])
+					droppedKeysAnonymousDefinitions[anonymized] = droppedKeys[1:]
+					renameIndex := &sqlparser.RenameIndex{
+						OldName: dropKey.Name,
+						NewName: t2Key.Info.Name,
+					}
+					alterTable.AlterOptions = append(alterTable.AlterOptions, renameIndex)
+					convertedToRename = true
+				}
+			}
+			if convertedToRename {
+				continue
+			}
+			// End of conversion to RENAME INDEX. Proceed with actual ADD INDEX
+
 			addKey := &sqlparser.AddIndexDefinition{
 				IndexDefinition: t2Key,
 			}
@@ -1663,6 +1711,9 @@ func (c *CreateTableEntity) diffKeys(alterTable *sqlparser.AlterTable,
 			}
 		}
 	}
+	for _, stmt := range dropKeyStatements {
+		alterTable.AlterOptions = append(alterTable.AlterOptions, stmt)
+	}
 	return superfluousFulltextKeys
 }
 
@@ -2036,12 +2087,14 @@ func sortAlterOptions(diff *AlterTableEntityDiff) {
 			return 5
 		case *sqlparser.AddColumns:
 			return 6
-		case *sqlparser.AddIndexDefinition:
+		case *sqlparser.RenameIndex:
 			return 7
-		case *sqlparser.AddConstraintDefinition:
+		case *sqlparser.AddIndexDefinition:
 			return 8
-		case sqlparser.TableOptions, *sqlparser.TableOptions:
+		case *sqlparser.AddConstraintDefinition:
 			return 9
+		case sqlparser.TableOptions, *sqlparser.TableOptions:
+			return 10
 		default:
 			return math.MaxInt
 		}
@@ -2197,20 +2250,25 @@ func (c *CreateTableEntity) apply(diff *AlterTableEntityDiff) error {
 				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.RenameIndex:
+			// validate no existing key by same name
+			newKeyName := opt.NewName.String()
+			for _, index := range c.TableSpec.Indexes {
+				if strings.EqualFold(index.Info.Name.String(), newKeyName) {
+					return &ApplyDuplicateKeyError{Table: c.Name(), Key: newKeyName}
 				}
 			}
-
+			found := false
+			for _, index := range c.TableSpec.Indexes {
+				if index.Info.Name.String() == opt.OldName.String() {
+					index.Info.Name = opt.NewName
+					found = true
+					break
+				}
+			}
+			if !found {
+				return &ApplyKeyNotFoundError{Table: c.Name(), Key: opt.OldName.String()}
+			}
 		case *sqlparser.AddIndexDefinition:
 			// validate no existing key by same name
 			keyName := opt.IndexDefinition.Info.Name.String()
@@ -2407,11 +2465,41 @@ func (c *CreateTableEntity) apply(diff *AlterTableEntityDiff) error {
 		}
 		return nil
 	}
+	// postApplyOptionsIteration runs on all options, after applyAlterOption does.
+	// Some validations can only take place after all options have been applied.
+	postApplyOptionsIteration := func(opt sqlparser.AlterOption) error {
+		switch opt := opt.(type) {
+		case *sqlparser.DropKey:
+			// Now, if this is a normal key being dropped, let's validate it does not leave any foreign key constraint uncovered.
+			// We must have this in `postApplyOptionsIteration` as opposed to `applyAlterOption` because
+			// this DROP KEY may have been followed by an ADD KEY that covers the foreign key constraint, so it's wrong
+			// to error out before applying the ADD KEY.
+			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()}
+					}
+				}
+			}
+		}
+		return nil
+	}
+
 	for _, alterOption := range diff.alterTable.AlterOptions {
 		if err := applyAlterOption(alterOption); err != nil {
 			return err
 		}
 	}
+	for _, alterOption := range diff.alterTable.AlterOptions {
+		if err := postApplyOptionsIteration(alterOption); err != nil {
+			return err
+		}
+	}
 	if err := c.postApplyNormalize(); err != nil {
 		return err
 	}

go/vt/schemadiff/table_test.go

@@ -805,6 +805,41 @@ func TestCreateTableDiff(t *testing.T) {
 				"+	KEY `i_idx3` (`id`)",
 			},
 		},
+		{
+			name:  "reordered and renamed key",
+			from:  "create table t1 (`id` int primary key, i int, key i_idx(i), key i2_idx(i, `id`))",
+			to:    "create table t2 (`id` int primary key, i int, key i2_alternative (`i`, id), key i_idx ( i ) )",
+			diff:  "alter table t1 rename index i2_idx to i2_alternative",
+			cdiff: "ALTER TABLE `t1` RENAME INDEX `i2_idx` TO `i2_alternative`",
+		},
+		{
+			name:  "reordered and renamed keys",
+			from:  "create table t1 (`id` int primary key, i int, key i_idx(i), key i2_idx(i, `id`))",
+			to:    "create table t2 (`id` int primary key, i int, key i2_alternative (`i`, id), key i_alternative ( i ) )",
+			diff:  "alter table t1 rename index i2_idx to i2_alternative, rename index i_idx to i_alternative",
+			cdiff: "ALTER TABLE `t1` RENAME INDEX `i2_idx` TO `i2_alternative`, RENAME INDEX `i_idx` TO `i_alternative`",
+		},
+		{
+			name:  "multiple similar keys, one rename",
+			from:  "create table t1 (`id` int primary key, i int, key i_idx(i), key i2_idx(i))",
+			to:    "create table t2 (`id` int primary key, i int, key i_idx(i), key i2_alternative(i))",
+			diff:  "alter table t1 rename index i2_idx to i2_alternative",
+			cdiff: "ALTER TABLE `t1` RENAME INDEX `i2_idx` TO `i2_alternative`",
+		},
+		{
+			name:  "multiple similar keys, two renames",
+			from:  "create table t1 (`id` int primary key, i int, key i_idx(i), key i2_idx(i))",
+			to:    "create table t2 (`id` int primary key, i int, key i_alternative(i), key i2_alternative(i))",
+			diff:  "alter table t1 rename index i_idx to i_alternative, rename index i2_idx to i2_alternative",
+			cdiff: "ALTER TABLE `t1` RENAME INDEX `i_idx` TO `i_alternative`, RENAME INDEX `i2_idx` TO `i2_alternative`",
+		},
+		{
+			name:  "multiple similar keys, two renames, reorder",
+			from:  "create table t1 (`id` int primary key, i int, key i0 (i, id), key i_idx(i), key i2_idx(i))",
+			to:    "create table t2 (`id` int primary key, i int, key i_alternative(i), key i2_alternative(i), key i0 (i, id))",
+			diff:  "alter table t1 rename index i_idx to i_alternative, rename index i2_idx to i2_alternative",
+			cdiff: "ALTER TABLE `t1` RENAME INDEX `i_idx` TO `i_alternative`, RENAME INDEX `i2_idx` TO `i2_alternative`",
+		},
 		{
 			name:  "key made visible",
 			from:  "create table t1 (`id` int primary key, i int, key i_idx(i) invisible)",
@@ -2788,6 +2823,18 @@ func TestValidate(t *testing.T) {
 			alter:     "alter table t drop key `i`",
 			expectErr: &IndexNeededByForeignKeyError{Table: "t", Key: "i"},
 		},
+		{
+			name:  "allow drop key when also adding a different index for foreign key constraint",
+			from:  "create table t (id int primary key, i int, key i_idx (i), constraint f foreign key (i) references parent(id))",
+			alter: "alter table t drop key `i_idx`, add key i_alternative (i)",
+			to:    "create table t (id int primary key, i int, key i_alternative (i), constraint f foreign key (i) references parent(id))",
+		},
+		{
+			name:  "allow drop key when also adding a different, longer, index for foreign key constraint",
+			from:  "create table t (id int primary key, i int, key i_idx (i), constraint f foreign key (i) references parent(id))",
+			alter: "alter table t drop key `i_idx`, add key i_alternative (i, id)",
+			to:    "create table t (id int primary key, i int, key i_alternative (i, id), constraint f foreign key (i) references parent(id))",
+		},
 		{
 			name:  "drop key with alternative key for foreign key constraint, 1",
 			from:  "create table t (id int primary key, i int, key i (i), key i2 (i, id), constraint f foreign key (i) references parent(id))",