Cherry-pick 47e1375 with conflicts

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

@@ -108,6 +108,14 @@ const (
 	AlterTableAlgorithmStrategyCopy
 )
 
+<<<<<<< HEAD
+=======
+const (
+	EnumReorderStrategyAllow int = iota
+	EnumReorderStrategyReject
+)
+
+>>>>>>> 47e137539b (VReplication/OnlineDDL: reordering enum values (#15103))
 // DiffHints is an assortment of rules for diffing entities
 type DiffHints struct {
 	StrictIndexOrdering         bool

go/vt/vttablet/onlineddl/executor.go

@@ -2625,7 +2625,14 @@ func (e *Executor) evaluateDeclarativeDiff(ctx context.Context, onlineDDL *schem
 	if newShowCreateTable == "" {
 		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected: cannot find table or view even as it was just created: %v", onlineDDL.Table)
 	}
+<<<<<<< HEAD
 	hints := &schemadiff.DiffHints{AutoIncrementStrategy: schemadiff.AutoIncrementApplyHigher}
+=======
+	senv := schemadiff.NewEnv(e.env.Environment(), e.env.Environment().CollationEnv().DefaultConnectionCharset())
+	hints := &schemadiff.DiffHints{
+		AutoIncrementStrategy: schemadiff.AutoIncrementApplyHigher,
+	}
+>>>>>>> 47e137539b (VReplication/OnlineDDL: reordering enum values (#15103))
 	switch ddlStmt.(type) {
 	case *sqlparser.CreateTable:
 		diff, err = schemadiff.DiffCreateTablesQueries(existingShowCreateTable, newShowCreateTable, hints)

go/vt/vttablet/onlineddl/vrepl.go

@@ -434,11 +434,26 @@ func (v *VRepl) analyzeTables(ctx context.Context, conn *dbconnpool.DBConnection
 	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)
+		})
+	}
+}