schemadiff: fix index/foreign-key apply scenario (#16698)

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>

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))",