ddl: fix rename partition column panic (#38932) (#38989)

close #38932
pingcap · Nov 9, 2022 · 9efcb88 · 9efcb88
1 parent ce62915
commit 9efcb88
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 1 deletion.
diff --git a/ddl/db_partition_test.go b/ddl/db_partition_test.go
@@ -4535,6 +4535,7 @@ func TestPartitionTableWithAnsiQuotes(t *testing.T) {
 		` PARTITION "p4" VALUES LESS THAN ('\\''\t\n','\\''\t\n'),` + "\n" +
 		` PARTITION "pMax" VALUES LESS THAN (MAXVALUE,MAXVALUE))`))
 }
+
 func TestAlterModifyColumnOnPartitionedTable(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
@@ -4665,6 +4666,28 @@ func TestAlterModifyColumnOnPartitionedTable(t *testing.T) {
 	tk.MustGetErrCode(`alter table t modify a varchar(20)`, errno.ErrUnsupportedDDLOperation)
 }
 
+func TestAlterModifyColumnOnPartitionedTableRename(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	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.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) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)

diff --git a/ddl/ddl_api.go b/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

diff --git a/ddl/partition.go b/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