Fix panic if partition column was renamed (pingcap#38932)

ddl/db_partition_test.go

@@ -4672,8 +4672,20 @@ func TestAlterModifyColumnOnPartitionedTableRename(t *testing.T) {
 	schemaName := "modColPartRename"
 	tk.MustExec("create database " + schemaName)
 	tk.MustExec("use " + schemaName)
+	tk.MustExec(`create table t (a int, b char) partition by range (a) (partition p0 values less than (10))`)
+	tk.MustContainErrMsg(`alter table t change a c int`, "[planner:1054]Unknown column 'a' in 'expression'")
+	tk.MustExec(`drop table t`)
 	tk.MustExec(`create table t (a char, b char) partition by range columns (a) (partition p0 values less than ('z'))`)
-	tk.MustExec(`alter table t change a c char`)
+	tk.MustContainErrMsg(`alter table t change a c char`, "[ddl:8200]New column does not match partition definitions: [ddl:1567]partition column name cannot be found")
+	tk.MustExec(`drop table t`)
+	tk.MustExec(`create table t (a int, b char) partition by list (a) (partition p0 values in (10))`)
+	tk.MustContainErrMsg(`alter table t change a c int`, "[planner:1054]Unknown column 'a' in 'expression'")
+	tk.MustExec(`drop table t`)
+	tk.MustExec(`create table t (a char, b char) partition by list columns (a) (partition p0 values in ('z'))`)
+	tk.MustContainErrMsg(`alter table t change a c char`, "[ddl:8200]New column does not match partition definitions: [ddl:1567]partition column name cannot be found")
+	tk.MustExec(`drop table t`)
+	tk.MustExec(`create table t (a int, b char) partition by hash (a) partitions 3`)
+	tk.MustContainErrMsg(`alter table t change a c int`, "[planner:1054]Unknown column 'a' in 'expression'")
 }
 
 func TestAlterModifyColumnOnPartitionedTableFail(t *testing.T) {

ddl/ddl_api.go

@@ -4670,6 +4670,7 @@ func GetModifiableColumnJob(
 			newTblInfo.Columns = newCols
 
 			var buf bytes.Buffer
+			// TODO: update the partitioning columns with new names if column is renamed
 			AppendPartitionInfo(tblInfo.GetPartitionInfo(), &buf, mysql.ModeNone)
 			// The parser supports ALTER TABLE ... PARTITION BY ... even if the ddl code does not yet :)
 			// Ignoring warnings

ddl/partition.go

@@ -782,6 +782,9 @@ func generatePartitionDefinitionsFromInterval(ctx sessionctx.Context, partOption
 	}
 	if len(tbInfo.Partition.Columns) > 0 {
 		colTypes := collectColumnsType(tbInfo)
+		if len(colTypes) != len(tbInfo.Partition.Columns) {
+			return dbterror.ErrWrongPartitionName.GenWithStack("partition column name cannot be found")
+		}
 		if _, err := checkAndGetColumnsTypeAndValuesMatch(ctx, colTypes, first.Exprs); err != nil {
 			return err
 		}
@@ -1081,6 +1084,9 @@ func buildListPartitionDefinitions(ctx sessionctx.Context, defs []*ast.Partition
 	definitions := make([]model.PartitionDefinition, 0, len(defs))
 	exprChecker := newPartitionExprChecker(ctx, nil, checkPartitionExprAllowed)
 	colTypes := collectColumnsType(tbInfo)
+	if len(colTypes) != len(tbInfo.Partition.Columns) {
+		return nil, dbterror.ErrWrongPartitionName.GenWithStack("partition column name cannot be found")
+	}
 	for _, def := range defs {
 		if err := def.Clause.Validate(model.PartitionTypeList, len(tbInfo.Partition.Columns)); err != nil {
 			return nil, err
@@ -1139,7 +1145,11 @@ func collectColumnsType(tbInfo *model.TableInfo) []types.FieldType {
 	if len(tbInfo.Partition.Columns) > 0 {
 		colTypes := make([]types.FieldType, 0, len(tbInfo.Partition.Columns))
 		for _, col := range tbInfo.Partition.Columns {
-			colTypes = append(colTypes, findColumnByName(col.L, tbInfo).FieldType)
+			c := findColumnByName(col.L, tbInfo)
+			if c == nil {
+				return nil
+			}
+			colTypes = append(colTypes, c.FieldType)
 		}
 
 		return colTypes
@@ -1152,6 +1162,9 @@ func buildRangePartitionDefinitions(ctx sessionctx.Context, defs []*ast.Partitio
 	definitions := make([]model.PartitionDefinition, 0, len(defs))
 	exprChecker := newPartitionExprChecker(ctx, nil, checkPartitionExprAllowed)
 	colTypes := collectColumnsType(tbInfo)
+	if len(colTypes) != len(tbInfo.Partition.Columns) {
+		return nil, dbterror.ErrWrongPartitionName.GenWithStack("partition column name cannot be found")
+	}
 	for _, def := range defs {
 		if err := def.Clause.Validate(model.PartitionTypeRange, len(tbInfo.Partition.Columns)); err != nil {
 			return nil, err