Improve dealing with check constraints (#10268)

This fixes a number of issues for check constaints. First it allows
parsing both `DROP CONSTRAINT` and `DROP CHECK` in the parser so we can
correctly drop an existing constraint. This was missing so far and only
a foreign key constraint could be dropped.

Secondly, we improve dealing with constraints in schemadiff as well.
Normalization there needs to generate names for constraints much like
how we generate names for keys as well.

We also need to teach the diffing logic then how to create the proper
`DROP` statements instead of a `DROP FOREIGN KEY` for all constraints
that get dropped.

Signed-off-by: Dirkjan Bussink <d.bussink@gmail.com>

go/vt/schemadiff/diff_test.go

@@ -340,6 +340,17 @@ func TestDiffSchemas(t *testing.T) {
 				"ALTER TABLE `t1` DROP KEY `deleted_check`, ADD UNIQUE KEY `deleted_check` (`id`, (if(`deleted_at` IS NOT NULL, 0, NULL)))",
 			},
 		},
+		{
+			name: "change for a check",
+			from: "CREATE TABLE `t` (`id` int NOT NULL, `test` int NOT NULL DEFAULT '0', PRIMARY KEY (`id`), CONSTRAINT `Check1` CHECK ((`test` >= 0)))",
+			to:   "CREATE TABLE `t` (`id` int NOT NULL, `test` int NOT NULL DEFAULT '0', PRIMARY KEY (`id`), CONSTRAINT `RenamedCheck1` CHECK ((`test` >= 0)))",
+			diffs: []string{
+				"alter table t drop check Check1, add constraint RenamedCheck1 check (test >= 0)",
+			},
+			cdiffs: []string{
+				"ALTER TABLE `t` DROP CHECK `Check1`, ADD CONSTRAINT `RenamedCheck1` CHECK (`test` >= 0)",
+			},
+		},
 		{
 			name: "change of table columns, removed",
 			from: "create table t(id int primary key, i int)",

go/vt/schemadiff/table.go

@@ -220,6 +220,7 @@ func NewCreateTableEntity(c *sqlparser.CreateTable) *CreateTableEntity {
 // The function returns this receiver as courtesy
 func (c *CreateTableEntity) normalize() *CreateTableEntity {
 	c.normalizeUnnamedKeys()
+	c.normalizeUnnamedConstraints()
 	c.normalizeTableOptions()
 	c.normalizeColumnOptions()
 	c.normalizePartitionOptions()
@@ -403,7 +404,7 @@ func (c *CreateTableEntity) normalizePartitionOptions() {
 func (c *CreateTableEntity) normalizeUnnamedKeys() {
 	// let's ensure all keys have names
 	keyNameExists := map[string]bool{}
-	// first, we iterate and take note for all keys that do aleady have names
+	// first, we iterate and take note for all keys that do already have names
 	for _, key := range c.CreateTable.TableSpec.Indexes {
 		if name := key.Info.Name.String(); name != "" {
 			keyNameExists[name] = true
@@ -441,6 +442,35 @@ func (c *CreateTableEntity) normalizeUnnamedKeys() {
 	}
 }
 
+func (c *CreateTableEntity) normalizeUnnamedConstraints() {
+	// let's ensure all keys 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 {
+		if name := constraint.Name.String(); name != "" {
+			constraintNameExists[name] = true
+		}
+	}
+
+	// now, let's look at keys that do not have names, and assign them new names
+	for _, constraint := range c.CreateTable.TableSpec.Constraints {
+		if name := constraint.Name.String(); name == "" {
+			nameFormat := "%s_chk_%d"
+			if _, fk := constraint.Details.(*sqlparser.ForeignKeyDefinition); fk {
+				nameFormat = "%s_ibfk_%d"
+			}
+			suggestedCheckName := fmt.Sprintf(nameFormat, c.CreateTable.Table.Name.String(), 1)
+			// now let's see if that name is taken; if it is, enumerate new news until we find a free name
+			for enumerate := 2; constraintNameExists[suggestedCheckName]; enumerate++ {
+				suggestedCheckName = fmt.Sprintf(nameFormat, c.CreateTable.Table.Name.String(), enumerate)
+			}
+			// OK we found a free slot!
+			constraint.Name = sqlparser.NewColIdent(suggestedCheckName)
+			constraintNameExists[suggestedCheckName] = true
+		}
+	}
+}
+
 // Name implements Entity interface
 func (c *CreateTableEntity) Name() string {
 	return c.CreateTable.GetTable().Name.String()
@@ -512,7 +542,7 @@ func (c *CreateTableEntity) TableDiff(other *CreateTableEntity, hints *DiffHints
 		c.diffKeys(alterTable, t1Keys, t2Keys, hints)
 	}
 	{
-		// diff constraints (foreign keys)
+		// diff constraints
 		// ordered constraints for both tables:
 		t1Constraints := c.CreateTable.TableSpec.Constraints
 		t2Constraints := other.CreateTable.TableSpec.Constraints
@@ -907,16 +937,19 @@ func (c *CreateTableEntity) diffConstraints(alterTable *sqlparser.AlterTable,
 		t2ConstraintsMap[constraint.Name.String()] = constraint
 	}
 
-	dropConstraintStatement := func(name sqlparser.ColIdent) *sqlparser.DropKey {
-		return &sqlparser.DropKey{Name: name, Type: sqlparser.ForeignKeyType}
+	dropConstraintStatement := func(constraint *sqlparser.ConstraintDefinition) *sqlparser.DropKey {
+		if _, fk := constraint.Details.(*sqlparser.ForeignKeyDefinition); fk {
+			return &sqlparser.DropKey{Name: constraint.Name, Type: sqlparser.ForeignKeyType}
+		}
+		return &sqlparser.DropKey{Name: constraint.Name, Type: sqlparser.CheckKeyType}
 	}
 
 	// evaluate dropped constraints
 	//
 	for _, t1Constraint := range t1Constraints {
 		if _, ok := t2ConstraintsMap[t1Constraint.Name.String()]; !ok {
 			// column exists in t1 but not in t2, hence it is dropped
-			dropConstraint := dropConstraintStatement(t1Constraint.Name)
+			dropConstraint := dropConstraintStatement(t1Constraint)
 			alterTable.AlterOptions = append(alterTable.AlterOptions, dropConstraint)
 		}
 	}
@@ -931,7 +964,7 @@ func (c *CreateTableEntity) diffConstraints(alterTable *sqlparser.AlterTable,
 			if sqlparser.CanonicalString(t2Constraint) != sqlparser.CanonicalString(t1Constraint) {
 				// constraints with same name have different definition. There is no ALTER INDEX statement,
 				// we're gonna drop and create.
-				dropConstraint := dropConstraintStatement(t1Constraint.Name)
+				dropConstraint := dropConstraintStatement(t1Constraint)
 				addConstraint := &sqlparser.AddConstraintDefinition{
 					ConstraintDefinition: t2Constraint,
 				}
@@ -1324,7 +1357,7 @@ func (c *CreateTableEntity) apply(diff *AlterTableEntityDiff) error {
 						break
 					}
 				}
-			case sqlparser.ForeignKeyType:
+			case sqlparser.ForeignKeyType, sqlparser.CheckKeyType:
 				for i, constraint := range c.TableSpec.Constraints {
 					if constraint.Name.String() == opt.Name.String() {
 						found = true

go/vt/schemadiff/table_test.go

@@ -1104,7 +1104,6 @@ func TestNormalize(t *testing.T) {
 			from: "create table if not exists t (id int primary key, i int)",
 			to:   "CREATE TABLE IF NOT EXISTS `t` (\n\t`id` int PRIMARY KEY,\n\t`i` int\n)",
 		},
-
 		{
 			name: "timestamp null",
 			from: "create table t (id int primary key, t timestamp null)",
@@ -1220,6 +1219,16 @@ func TestNormalize(t *testing.T) {
 			from: "create table a (id int not null primary key) engine InnoDB, charset utf8mb4, collate utf8mb4_0900_ai_ci partition by range (`id`) (partition `p10` values less than(10) engine innodb)",
 			to:   "CREATE TABLE `a` (\n\t`id` int NOT NULL PRIMARY KEY\n) ENGINE InnoDB,\n  CHARSET utf8mb4,\n  COLLATE utf8mb4_0900_ai_ci PARTITION BY RANGE (`id`) (PARTITION `p10` VALUES LESS THAN (10) ENGINE InnoDB)",
 		},
+		{
+			name: "generates a name for checks",
+			from: "create table t (id int NOT NULL, test int NOT NULL DEFAULT 0, PRIMARY KEY (id), CHECK ((test >= 0)))",
+			to:   "CREATE TABLE `t` (\n\t`id` int NOT NULL,\n\t`test` int NOT NULL DEFAULT 0,\n\tPRIMARY KEY (`id`),\n\tCONSTRAINT `t_chk_1` CHECK (`test` >= 0)\n)",
+		},
+		{
+			name: "generates a name for foreign key constraints",
+			from: "create table t1 (id int primary key, i int, foreign key (i) references parent(id))",
+			to:   "CREATE TABLE `t1` (\n\t`id` int PRIMARY KEY,\n\t`i` int,\n\tCONSTRAINT `t1_ibfk_1` FOREIGN KEY (`i`) REFERENCES `parent` (`id`)\n)",
+		},
 	}
 	for _, ts := range tt {
 		t.Run(ts.name, func(t *testing.T) {

go/vt/sqlparser/ast_funcs.go

@@ -1611,6 +1611,8 @@ func (key DropKeyType) ToString() string {
 		return ForeignKeyTypeStr
 	case NormalKeyType:
 		return NormalKeyTypeStr
+	case CheckKeyType:
+		return CheckKeyTypeStr
 	default:
 		return "Unknown DropKeyType"
 	}

go/vt/sqlparser/constants.go

@@ -302,6 +302,7 @@ const (
 	PrimaryKeyTypeStr = "primary key"
 	ForeignKeyTypeStr = "foreign key"
 	NormalKeyTypeStr  = "key"
+	CheckKeyTypeStr   = "check"
 
 	// TrimType strings
 	BothTrimStr     = "both"
@@ -682,6 +683,7 @@ const (
 	PrimaryKeyType DropKeyType = iota
 	ForeignKeyType
 	NormalKeyType
+	CheckKeyType
 )
 
 // LockOptionType constants

go/vt/sqlparser/parse_test.go

@@ -1283,6 +1283,8 @@ var (
 	}, {
 		input:  "alter table a add constraint check (id)",
 		output: "alter table a add check (id)",
+	}, {
+		input: "alter table a add constraint c check (id)",
 	}, {
 		input:  "alter table a add id int",
 		output: "alter table a add column id int",
@@ -1300,17 +1302,14 @@ var (
 	}, {
 		input: "alter table a add check (ch_1) not enforced",
 	}, {
-		input:      "alter table a drop check ch_1",
-		output:     "alter table a",
-		partialDDL: true,
+		input: "alter table a drop check ch_1",
+	}, {
+		input:  "alter table a drop constraint ch_1",
+		output: "alter table a drop check ch_1",
 	}, {
 		input: "alter table a drop foreign key kx",
 	}, {
 		input: "alter table a drop primary key",
-	}, {
-		input:      "alter table a drop constraint",
-		output:     "alter table a",
-		partialDDL: true,
 	}, {
 		input:  "alter table a drop id",
 		output: "alter table a drop column id",