schemadiff: additional FK column type matching rules (#14751)

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

Author: shlomi-noach

go/vt/schemadiff/schema.go

@@ -350,12 +350,26 @@ func (s *Schema) normalize() error {
 			return errors.Join(errs, err)
 		}
 	}
-	colTypeEqualForForeignKey := func(a, b *sqlparser.ColumnType) bool {
-		return a.Type == b.Type &&
-			a.Unsigned == b.Unsigned &&
-			a.Zerofill == b.Zerofill &&
-			sqlparser.Equals.ColumnCharset(a.Charset, b.Charset) &&
-			sqlparser.Equals.SliceOfString(a.EnumValues, b.EnumValues)
+	colTypeCompatibleForForeignKey := func(child, parent *sqlparser.ColumnType) bool {
+		if child.Type == parent.Type {
+			return true
+		}
+		if child.Type == "char" && parent.Type == "varchar" {
+			return true
+		}
+		return false
+	}
+	colTypeEqualForForeignKey := func(child, parent *sqlparser.ColumnType) bool {
+		if colTypeCompatibleForForeignKey(child, parent) &&
+			child.Unsigned == parent.Unsigned &&
+			child.Zerofill == parent.Zerofill &&
+			sqlparser.Equals.ColumnCharset(child.Charset, parent.Charset) &&
+			child.Options.Collate == parent.Options.Collate &&
+			sqlparser.Equals.SliceOfString(child.EnumValues, parent.EnumValues) {
+			// Complete identify (other than precision which is ignored)
+			return true
+		}
+		return false
 	}
 
 	// Now validate foreign key columns:

go/vt/schemadiff/schema_test.go

@@ -378,10 +378,31 @@ func TestInvalidSchema(t *testing.T) {
 			// InnoDB allows different string length
 			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)",
 		},
+		{
+			// explicit charset/collation
+			schema: "create table t10(id varchar(50) charset utf8mb4 collate utf8mb4_0900_ai_ci primary key); create table t11 (id int primary key, i varchar(100) charset utf8mb4 collate utf8mb4_0900_ai_ci, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
+		},
+		{
+			// weirdly allowed: varchar->char
+			schema: "create table t10(id varchar(50) charset utf8mb4 collate utf8mb4_0900_ai_ci primary key); create table t11 (id int primary key, i char(100) charset utf8mb4 collate utf8mb4_0900_ai_ci, key ix(i), constraint f10 foreign key (i) references t10(id) on delete restrict)",
+		},
+		{
+			// but weirdly not allowed: char->varchar
+			schema:    "create table t10(id char(50) charset utf8mb4 collate utf8mb4_0900_ai_ci primary key); create table t11 (id int primary key, i varchar(50) charset utf8mb4 collate utf8mb4_0900_ai_ci, 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 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"},
 		},
+		{
+			schema:    "create table t10(id varchar(50) charset utf8mb4 collate utf8mb4_0900_ai_ci primary key); create table t11 (id int primary key, i varchar(100) charset utf8mb4 collate utf8mb4_general_ci, 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 VARCHAR(50) charset utf8mb4 collate utf8mb4_0900_ai_ci primary key); create table t11 (id int primary key, i VARCHAR(100) charset utf8mb4 collate utf8mb4_general_ci, 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"},
+		},
 	}
 	for _, ts := range tt {
 		t.Run(ts.schema, func(t *testing.T) {