ddl: check index key length when convert charset (#56964)

close #56930

pkg/ddl/executor.go

@@ -51,6 +51,7 @@ import (
 	pmodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/terror"
+	parser_types "github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tidb/pkg/privilege"
 	rg "github.com/pingcap/tidb/pkg/resourcegroup"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -3962,6 +3963,10 @@ func checkAlterTableCharset(tblInfo *model.TableInfo, dbInfo *model.DBInfo, toCh
 		return doNothing, nil
 	}
 
+	if err = checkIndexLengthWithNewCharset(tblInfo, toCharset, toCollate); err != nil {
+		return doNothing, err
+	}
+
 	for _, col := range tblInfo.Columns {
 		if col.GetType() == mysql.TypeVarchar {
 			if err = types.IsVarcharTooBigFieldLength(col.GetFlen(), col.Name.O, toCharset); err != nil {
@@ -3985,6 +3990,30 @@ func checkAlterTableCharset(tblInfo *model.TableInfo, dbInfo *model.DBInfo, toCh
 	return doNothing, nil
 }
 
+func checkIndexLengthWithNewCharset(tblInfo *model.TableInfo, toCharset, toCollate string) error {
+	// Copy all columns and replace the charset and collate.
+	columns := make([]*model.ColumnInfo, 0, len(tblInfo.Columns))
+	for _, col := range tblInfo.Columns {
+		newCol := col.Clone()
+		if parser_types.HasCharset(&newCol.FieldType) {
+			newCol.SetCharset(toCharset)
+			newCol.SetCollate(toCollate)
+		} else {
+			newCol.SetCharset(charset.CharsetBin)
+			newCol.SetCollate(charset.CharsetBin)
+		}
+		columns = append(columns, newCol)
+	}
+
+	for _, indexInfo := range tblInfo.Indices {
+		err := checkIndexPrefixLength(columns, indexInfo.Columns)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // RenameIndex renames an index.
 // In TiDB, indexes are case-insensitive (so index 'a' and 'A" are considered the same index),
 // but index names are case-sensitive (we can rename index 'a' to 'A')

tests/integrationtest/r/session/common.result

@@ -137,6 +137,18 @@ create table t (a blob(10000), b timestamp, index idx(a(3069), b));
 Error 1071 (42000): Specified key was too long (3073 bytes); max key length is 3072 bytes
 create table t (a blob(10000), b timestamp, index idx(a(3068), b));
 drop table if exists t;
+create table posts (id int auto_increment primary key, title varchar(500) character set utf8, subtitle varchar(500) character set utf8, unique key(title, subtitle));
+alter table posts convert to character set utf8mb4;
+Error 1071 (42000): Specified key was too long (4000 bytes); max key length is 3072 bytes
+drop table if exists posts;
+create table t(a varchar(1000) character set utf8, primary key(a));
+alter table t convert to character set utf8mb4;
+Error 1071 (42000): Specified key was too long (4000 bytes); max key length is 3072 bytes
+drop table if exists t;
+create table t(a varchar(1000) character set utf8, key(a));
+alter table t convert to character set utf8mb4;
+Error 1071 (42000): Specified key was too long (4000 bytes); max key length is 3072 bytes
+drop table if exists t;
 drop table if exists t1; create table t1(id int ); insert into t1 values (1);
 select * from t1;
 id

tests/integrationtest/t/session/common.test

@@ -140,6 +140,18 @@ drop table if exists t;
 create table t (a blob(10000), b timestamp, index idx(a(3069), b));
 create table t (a blob(10000), b timestamp, index idx(a(3068), b));
 drop table if exists t;
+create table posts (id int auto_increment primary key, title varchar(500) character set utf8, subtitle varchar(500) character set utf8, unique key(title, subtitle));
+-- error 1071
+alter table posts convert to character set utf8mb4;
+drop table if exists posts;
+create table t(a varchar(1000) character set utf8, primary key(a));
+-- error 1071
+alter table t convert to character set utf8mb4;
+drop table if exists t;
+create table t(a varchar(1000) character set utf8, key(a));
+-- error 1071
+alter table t convert to character set utf8mb4;
+drop table if exists t;
 
 # TestMultiStmts
 drop table if exists t1; create table t1(id int ); insert into t1 values (1);