VReplication/OnlineDDL: reordering enum values

go/test/endtoend/onlineddl/vrepl_suite/testdata/enum-reorder/alter

@@ -0,0 +1 @@
+change e e enum('blue', 'green', 'red') not null

go/test/endtoend/onlineddl/vrepl_suite/testdata/enum-reorder/create.sql

@@ -0,0 +1,26 @@
+drop table if exists onlineddl_test;
+create table onlineddl_test (
+  id int auto_increment,
+  i int not null,
+  e enum('red', 'green', 'blue') not null,
+  primary key(id)
+) auto_increment=1;
+
+insert into onlineddl_test values (null, 11, 'red');
+insert into onlineddl_test values (null, 13, 'green');
+insert into onlineddl_test values (null, 17, 'blue');
+
+drop event if exists onlineddl_test;
+delimiter ;;
+create event onlineddl_test
+  on schedule every 1 second
+  starts current_timestamp
+  ends current_timestamp + interval 60 second
+  on completion not preserve
+  enable
+  do
+begin
+  insert into onlineddl_test values (null, 211, 'red');
+  insert into onlineddl_test values (null, 213, 'green');
+  insert into onlineddl_test values (null, 217, 'blue');
+end ;;

go/vt/schemadiff/types.go

@@ -120,8 +120,8 @@ const (
 )
 
 const (
-	EnumReorderStrategyReject int = iota
-	EnumReorderStrategyAllow
+	EnumReorderStrategyAllow int = iota
+	EnumReorderStrategyReject
 )
 
 // DiffHints is an assortment of rules for diffing entities

go/vt/vttablet/onlineddl/executor.go

@@ -2801,7 +2801,6 @@ func (e *Executor) evaluateDeclarativeDiff(ctx context.Context, onlineDDL *schem
 	senv := schemadiff.NewEnv(e.env.Environment(), e.env.Environment().CollationEnv().DefaultConnectionCharset())
 	hints := &schemadiff.DiffHints{
 		AutoIncrementStrategy: schemadiff.AutoIncrementApplyHigher,
-		EnumReorderStrategy:   schemadiff.EnumReorderStrategyAllow,
 	}
 	switch ddlStmt.(type) {
 	case *sqlparser.CreateTable:

go/vt/vttablet/onlineddl/vrepl.go

@@ -491,11 +491,26 @@
 	for i := range v.sourceSharedColumns.Columns() {
 		sourceColumn := v.sourceSharedColumns.Columns()[i]
 		mappedColumn := v.targetSharedColumns.Columns()[i]
-		if sourceColumn.Type == vrepl.EnumColumnType && mappedColumn.Type != vrepl.EnumColumnType && mappedColumn.Charset != "" {
-			// A column is converted from ENUM type to textual type
-			v.targetSharedColumns.SetEnumToTextConversion(mappedColumn.Name, sourceColumn.EnumValues)
-			v.enumToTextMap[sourceColumn.Name] = sourceColumn.EnumValues
+		if sourceColumn.Type == vrepl.EnumColumnType {
+			switch {
+			// Either this is an ENUM column that stays an ENUM, or it is converted to a textual type.
+			// We take note of the enum values, and make it available in vreplication's Filter.Rule.ConvertEnumToText.
+			// This, in turn, will be used by vplayer (in TablePlan) like so:
+			// - In the binary log, enum values are integers.
+			// - Upon seeing this map, PlanBuilder will convert said int to the enum's logical string value.
+			// - And will apply the value as a string (`StringBindVariable`) in the query.
+			// What this allows is for enum values to have different ordering in the before/after table schema,
+			// so that for example you could modify an enum column:
+			// - from `('red', 'green', 'blue')` to `('red', 'blue')`
+			// - from `('red', 'green', 'blue')` to `('blue', 'red', 'green')`
+			case mappedColumn.Type == vrepl.EnumColumnType:
+				v.enumToTextMap[sourceColumn.Name] = sourceColumn.EnumValues
+			case mappedColumn.Charset != "":
+				v.enumToTextMap[sourceColumn.Name] = sourceColumn.EnumValues
+				v.targetSharedColumns.SetEnumToTextConversion(mappedColumn.Name, sourceColumn.EnumValues)
+			}
 		}
+
 		if sourceColumn.IsIntegralType() && mappedColumn.Type == vrepl.EnumColumnType {
 			v.intToEnumMap[sourceColumn.Name] = true
 		}

go/vt/vttablet/onlineddl/vrepl/columns.go

@@ -129,7 +129,7 @@ func isExpandedColumn(sourceColumn *Column, targetColumn *Column) (bool, string)
 			return true, "source is unsigned, target is signed"
 		}
 		if sourceColumn.NumericPrecision <= targetColumn.NumericPrecision && !sourceColumn.IsUnsigned && targetColumn.IsUnsigned {
-			// e.g. INT SIGNED => INT UNSIGNED, INT SIGNED = BIGINT UNSIGNED
+			// e.g. INT SIGNED => INT UNSIGNED, INT SIGNED => BIGINT UNSIGNED
 			return true, "target unsigned value exceeds source unsigned value"
 		}
 		if targetColumn.IsFloatingPoint() && !sourceColumn.IsFloatingPoint() {

go/vt/vttablet/onlineddl/vrepl/columns_test.go

@@ -133,3 +133,248 @@ func TestGetSharedColumns(t *testing.T) {
 		})
 	}
 }
+
+func TestGetExpandedColumnNames(t *testing.T) {
+	var (
+		columnsA = &ColumnList{
+			columns: []Column{
+				{
+					Name:       "c1",
+					IsNullable: true,
+				},
+				{
+					Name:       "c2",
+					IsNullable: true,
+				},
+				{
+					Name:       "c3",
+					IsNullable: false,
+				},
+			},
+			Ordinals: ColumnsMap{},
+		}
+		columnsB = &ColumnList{
+			columns: []Column{
+				{
+					Name:       "c1",
+					IsNullable: true,
+				},
+				{
+					Name:       "c2",
+					IsNullable: false,
+				},
+				{
+					Name:       "c3",
+					IsNullable: true,
+				},
+			},
+			Ordinals: ColumnsMap{},
+		}
+	)
+	tcases := []struct {
+		name      string
+		sourceCol Column
+		targetCol Column
+		expanded  bool
+	}{
+		{
+			"both nullable",
+			Column{
+				IsNullable: true,
+			},
+			Column{
+				IsNullable: true,
+			},
+			false,
+		},
+		{
+			"nullable to non nullable",
+			Column{
+				IsNullable: true,
+			},
+			Column{
+				IsNullable: false,
+			},
+			false,
+		},
+		{
+			"non nullable to nullable",
+			Column{
+				IsNullable: false,
+			},
+			Column{
+				IsNullable: true,
+			},
+			true,
+		},
+		{
+			"signed to unsigned",
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 4,
+				IsUnsigned:       false,
+			},
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 4,
+				IsUnsigned:       true,
+			},
+			true,
+		},
+		{
+			"unsigned to signed",
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 4,
+				IsUnsigned:       true,
+			},
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 4,
+				IsUnsigned:       false,
+			},
+			true,
+		},
+		{
+			"signed to smaller unsigned",
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 8,
+				IsUnsigned:       false,
+			},
+			Column{
+				Type:             IntegerColumnType,
+				NumericPrecision: 4,
+				IsUnsigned:       true,
+			},
+			false,
+		},
+		{
+			"same char length",
+			Column{
+				CharacterMaximumLength: 20,
+			},
+			Column{
+				CharacterMaximumLength: 20,
+			},
+			false,
+		},
+		{
+			"reduced char length",
+			Column{
+				CharacterMaximumLength: 20,
+			},
+			Column{
+				CharacterMaximumLength: 19,
+			},
+			false,
+		},
+		{
+			"increased char length",
+			Column{
+				CharacterMaximumLength: 20,
+			},
+			Column{
+				CharacterMaximumLength: 21,
+			},
+			true,
+		},
+		{
+			"expand temporal",
+			Column{
+				DataType: "time",
+			},
+			Column{
+				DataType: "timestamp",
+			},
+			true,
+		},
+		{
+			"expand temporal",
+			Column{
+				DataType: "date",
+			},
+			Column{
+				DataType: "timestamp",
+			},
+			true,
+		},
+		{
+			"expand temporal",
+			Column{
+				DataType: "date",
+			},
+			Column{
+				DataType: "datetime",
+			},
+			true,
+		},
+		{
+			"non expand temporal",
+			Column{
+				DataType: "datetime",
+			},
+			Column{
+				DataType: "timestamp",
+			},
+			false,
+		},
+		{
+			"expand temporal",
+			Column{
+				DataType: "timestamp",
+			},
+			Column{
+				DataType: "datetime",
+			},
+			true,
+		},
+		{
+			"expand enum",
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'b'",
+			},
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'x'",
+			},
+			true,
+		},
+		{
+			"expand enum",
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'b'",
+			},
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'b', 'c'",
+			},
+			true,
+		},
+		{
+			"reduce enum",
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'b', 'c'",
+			},
+			Column{
+				Type:       EnumColumnType,
+				EnumValues: "'a', 'b'",
+			},
+			false,
+		},
+	}
+
+	expectedExpandedColumnNames := []string{"c3"}
+	expandedColumnNames, _ := GetExpandedColumnNames(columnsA, columnsB)
+	assert.Equal(t, expectedExpandedColumnNames, expandedColumnNames)
+
+	for _, tcase := range tcases {
+		t.Run(tcase.name, func(t *testing.T) {
+			expanded, _ := isExpandedColumn(&tcase.sourceCol, &tcase.targetCol)
+			assert.Equal(t, tcase.expanded, expanded)
+		})
+	}
+}

go/vt/vttablet/onlineddl/vrepl/foreign_key.go

@@ -38,7 +38,6 @@ func RemovedForeignKeyNames(
 	env := schemadiff.NewEnv(venv, venv.CollationEnv().DefaultConnectionCharset())
 	diffHints := schemadiff.DiffHints{
 		ConstraintNamesStrategy: schemadiff.ConstraintNamesIgnoreAll,
-		EnumReorderStrategy:     schemadiff.EnumReorderStrategyAllow,
 	}
 	diff, err := schemadiff.DiffCreateTablesQueries(env, originalCreateTable, vreplCreateTable, &diffHints)
 	if err != nil {

go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go

@@ -338,8 +338,13 @@ func (tp *TablePlan) bindFieldVal(field *querypb.Field, val *sqltypes.Value) (*q
 	if enumValues, ok := tp.EnumValuesMap[field.Name]; ok && !val.IsNull() {
 		// The fact that this field has a EnumValuesMap entry, means we must
 		// use the enum's text value as opposed to the enum's numerical value.
-		// Once known use case is with Online DDL, when a column is converted from
-		// ENUM to a VARCHAR/TEXT.
+		// This may be needed in Online DDL, when the enum column could be modified:
+		// - Either from ENUM to a text type (VARCHAR/TEXT)
+		// - Or from ENUM to another ENUM with different value ordering,
+		//   e.g. from `('red', 'green', 'blue')` to `('red', 'blue')`.
+		// By applying the textual value of an enum we eliminate the ordering concern.
+		// In non-Online DDL this shouldn't be a concern because the schema is static,
+		// and so passing the enum's numerical value is sufficient.
 		enumValue, enumValueOK := enumValues[val.ToString()]
 		if !enumValueOK {
 			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "Invalid enum value: %v for field %s", val, field.Name)