Skip to content

Commit

Permalink
Merge pull request #176 from hanchuanchuan/patch-index-length-limit-767
Browse files Browse the repository at this point in the history 
优化索引单列长度的767字节限制逻辑
  • Loading branch information
@hanchuanchuan
hanchuanchuan authored Mar 24, 2020
2 parents a92ce9d + 4dfd48d commit ce15c43
Show file tree
Hide file tree
Showing 5 changed files with 98 additions and 55 deletions.
39 changes: 22 additions & 17 deletions session/session_inception.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5196,7 +5196,7 @@ func (s *session) checkCreateIndex(table *ast.TableName, IndexName string,
keyMaxLen := 0
// 禁止使用blob列当索引,所以不再检测blob字段时列是否过长
isBlobColumn := false

isOverflowIndexLength := false
for _, col := range IndexColNames {
found := false
var foundField FieldInfo
Expand All @@ -5222,7 +5222,7 @@ func (s *session) checkCreateIndex(table *ast.TableName, IndexName string,
s.AppendErrorNo(ER_BLOB_USED_AS_KEY, foundField.Field)
}

maxLength := foundField.GetDataBytes(s.DBVersion, s.Inc.DefaultCharset)
columnIndexLength := foundField.GetDataBytes(s.DBVersion, s.Inc.DefaultCharset)

// Length must be specified for BLOB and TEXT column indexes.
// if types.IsTypeBlob(col.FieldType.Tp) && ic.Length == types.UnspecifiedLength {
Expand All @@ -5240,22 +5240,23 @@ func (s *session) checkCreateIndex(table *ast.TableName, IndexName string,
if (strings.Contains(strings.ToLower(foundField.Type), "blob") ||
strings.Contains(strings.ToLower(foundField.Type), "char") ||
strings.Contains(strings.ToLower(foundField.Type), "text")) &&
col.Length > maxLength {
col.Length > columnIndexLength {
s.AppendErrorNo(ER_WRONG_SUB_KEY)
col.Length = maxLength
col.Length = columnIndexLength
}
}

if col.Length == types.UnspecifiedLength {
keyMaxLen += maxLength
keyMaxLen += columnIndexLength
} else {
tmpField := &FieldInfo{
Field: foundField.Field,
Type: fmt.Sprintf("%s(%d)", GetDataTypeBase(foundField.Type), col.Length),
Collation: foundField.Collation,
}

keyMaxLen += tmpField.GetDataLength(s.DBVersion, s.Inc.DefaultCharset)
columnIndexLength = tmpField.GetDataLength(s.DBVersion, s.Inc.DefaultCharset)
keyMaxLen += columnIndexLength

// bysPerChar := 3
// charset := s.Inc.DefaultCharset
Expand All @@ -5274,6 +5275,13 @@ func (s *session) checkCreateIndex(table *ast.TableName, IndexName string,
// }
}

if !s.innodbLargePrefix && !isOverflowIndexLength &&
!isBlobColumn &&
columnIndexLength > maxKeyLength {
s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength)
isOverflowIndexLength = true
}

if tp == ast.ConstraintPrimaryKey {
fieldType := GetDataTypeBase(strings.ToLower(foundField.Type))

Expand All @@ -5299,21 +5307,18 @@ func (s *session) checkCreateIndex(table *ast.TableName, IndexName string,
s.AppendErrorMessage(fmt.Sprintf("表'%s'的索引'%s'名称过长", t.Name, IndexName))
}

if !isBlobColumn {
// mysqlVersion := s.DBVersion
// mysql 5.6版本索引长度限制是767,5.7及之后变为3072
// log.Info(s.innodbLargePrefix)
// log.Info(keyMaxLen)
if s.innodbLargePrefix && keyMaxLen > maxKeyLength57 {
if !isBlobColumn && !isOverflowIndexLength {
// --删除!-- mysql 5.6版本索引长度限制是767,5.7及之后变为3072
// 未开启innodbLargePrefix时,单列长度不能超过767
// 所有情况下,总长度不能超过3072
if keyMaxLen > maxKeyLength57 {
s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength57)
} else if !s.innodbLargePrefix && keyMaxLen > maxKeyLength {
s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength)
}

// if mysqlVersion < 50700 && keyMaxLen > maxKeyLength {
// s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength)
// } else if (mysqlVersion >= 50700) && keyMaxLen > maxKeyLength57 {
// if s.innodbLargePrefix && keyMaxLen > maxKeyLength57 {
// s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength57)
// } else if !s.innodbLargePrefix && keyMaxLen > maxKeyLength {
// s.AppendErrorNo(ER_TOO_LONG_KEY, IndexName, maxKeyLength)
// }
}

Expand Down
45 changes: 26 additions & 19 deletions session/session_inception_backup_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -380,6 +380,7 @@ func (s *testSessionIncBackupSuite) TestUpdate(c *C) {

sql = `update table1 t1,table2 t2 set t1.c1=10,t2.c22=20 where t1.id1=t2.id2 and t2.c1=1;`
s.mustRunBackup(c, sql)

s.assertRows(c, s.rows[1:],
"UPDATE `test_inc`.`table2` SET `id2`=1, `c1`=1, `c2`=1, `c22`=NULL WHERE `id2`=1;",
"UPDATE `test_inc`.`table2` SET `id2`=2, `c1`=1, `c2`=NULL, `c22`=NULL WHERE `id2`=2;",
Expand Down Expand Up @@ -714,7 +715,11 @@ func (s *testSessionIncBackupSuite) TestDelete(c *C) {
insert into t1 values(1,X'71E6D5A383BB447C');`)
s.runBackup("delete from t1;")
row = s.rows[int(s.session.AffectedRows())-1]
c.Assert(row[2], Equals, "2", Commentf("%v", row))
if s.DBVersion >= 50700 {
c.Assert(row[2], Equals, "2", Commentf("%v", row))
} else {
c.Assert(row[2], Equals, "0", Commentf("%v", row))
}
}

func (s *testSessionIncBackupSuite) TestCreateDataBase(c *C) {
Expand Down Expand Up @@ -789,33 +794,35 @@ func (s *testSessionIncBackupSuite) TestAlterTableDropIndex(c *C) {
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD UNIQUE INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))

sql = `drop table if exists t1;
// 几何类型字段低版本不支持
if s.DBVersion >= 50700 {
sql = `drop table if exists t1;
create table t1(id int primary key,c1 GEOMETRY not null ,SPATIAL index ix_1(c1));
alter table t1 drop index ix_1;`
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))

sql = `drop table if exists t1;
sql = `drop table if exists t1;
create table t1(id int primary key,c1 GEOMETRY not null);
alter table t1 add SPATIAL index ix_1(c1);
alter table t1 drop index ix_1;`
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))

sql = `drop table if exists t1;
sql = `drop table if exists t1;
create table t1(id int primary key,c1 GEOMETRY not null);
alter table t1 add SPATIAL index ix_1(c1);`
s.runBackup(sql)
sql = "alter table t1 drop index ix_1;"
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))

s.runBackup(sql)
sql = "alter table t1 drop index ix_1;"
s.mustRunBackup(c, sql)
row = s.rows[int(s.session.AffectedRows())-1]
backup = s.query("t1", row[7].(string))
c.Assert(backup, Equals, "ALTER TABLE `test_inc`.`t1` ADD SPATIAL INDEX `ix_1`(`c1`);", Commentf("%v", s.rows))
}
}

func (s *testSessionIncBackupSuite) TestAlterTable(c *C) {
Expand Down
42 changes: 31 additions & 11 deletions session/session_inception_common_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"fmt"
"path"
"runtime"
"sort"
"strconv"
"strings"
"testing"
Expand Down Expand Up @@ -138,6 +139,9 @@ func (s *testCommon) initSetUp(c *C) {
inc.SqlSafeUpdates = 0
inc.EnableDropTable = true

// mysql5.6测试用例会出错(docker映射对外的端口不一致)
config.GetGlobalConfig().Ghost.GhostAliyunRds = true

s.defaultInc = *inc

s.remoteBackupTable = "$_$Inception_backup_information$_$"
Expand Down Expand Up @@ -372,7 +376,7 @@ func (s *testCommon) mysqlServerVersion() error {
if err != nil {
return err
}
emptyInnodbLargePrefix := true
// emptyInnodbLargePrefix := true
for rows.Next() {
rows.Scan(&name, &value)

Expand All @@ -391,9 +395,9 @@ func (s *testCommon) mysqlServerVersion() error {
return errors.New(fmt.Sprintf("无法解析版本号:%s", value))
}
log.Debug("db version: ", s.DBVersion)
// case "innodb_large_prefix":
// emptyInnodbLargePrefix = false
// s.innodbLargePrefix = value == "ON"
case "innodb_large_prefix":
// emptyInnodbLargePrefix = false
s.innodbLargePrefix = value == "ON" || value == "1"
case "sql_mode":
s.sqlMode = value
case "lower_case_table_names":
Expand All @@ -410,13 +414,13 @@ func (s *testCommon) mysqlServerVersion() error {
}

// 如果没有innodb_large_prefix系统变量
if emptyInnodbLargePrefix {
if s.DBVersion > 50700 {
s.innodbLargePrefix = true
} else {
s.innodbLargePrefix = false
}
}
// if emptyInnodbLargePrefix {
// if s.DBVersion > 50700 {
// s.innodbLargePrefix = true
// } else {
// s.innodbLargePrefix = false
// }
// }
return nil
}

Expand Down Expand Up @@ -545,6 +549,22 @@ func (s *testCommon) assertRows(c *C, rows [][]interface{}, rollbackSqls ...stri

c.Assert(len(result), Equals, len(rollbackSqls), Commentf("%v", rows))

// 如果是UPDATE多表操作,此时回滚的SQL可能是无序的
if len(result) > 1 && strings.HasPrefix(result[0], "UPDATE") {
prefix := ""
for i, sql := range result {
if i == 0 {
prefix = strings.Fields(sql)[1]
continue
}
if prefix != strings.Fields(sql)[1] {
sort.Strings(result)
sort.Strings(rollbackSqls)
break
}
}
}

for i := range result {
c.Assert(result[i], Equals, rollbackSqls[i], Commentf("%v", result))
}
Expand Down
15 changes: 14 additions & 1 deletion session/session_inception_exec_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -268,7 +268,7 @@ func (s *testSessionIncExecSuite) TestCreateTable(c *C) {
fmt.Println("数据库版本: ", s.DBVersion)

indexMaxLength := 767
if s.DBVersion >= 50700 {
if s.innodbLargePrefix {
indexMaxLength = 3072
}

Expand All @@ -289,6 +289,19 @@ func (s *testSessionIncExecSuite) TestCreateTable(c *C) {
s.testErrorCode(c, sql,
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))

s.runExec("drop table if exists t1")
if s.innodbLargePrefix {
sql = "create table t1(c1 int,c2 varchar(400),c3 varchar(400),index idx_1(c1,c2))charset utf8;;"
s.testErrorCode(c, sql)
} else {
sql = "create table t1(c1 int,c2 varchar(400),c3 varchar(400),index idx_1(c1,c2))charset utf8;;"
s.testErrorCode(c, sql,
session.NewErr(session.ER_TOO_LONG_KEY, "idx_1", indexMaxLength))
}

sql = "drop table if exists t1;create table t1(c1 int,c2 varchar(200),c3 varchar(200),index idx_1(c1,c2))charset utf8;;"
s.testErrorCode(c, sql)

config.GetGlobalConfig().Inc.EnableBlobType = true
// sql = "drop table if exists t1;create table t1(c1 int,c2 text, unique uq_1(c1,c2(3068)));"
// if indexMaxLength == 3072 {
Expand Down
12 changes: 5 additions & 7 deletions session/session_inception_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -599,7 +599,7 @@ func (s *testSessionIncSuite) TestCreateTable(c *C) {
config.GetGlobalConfig().Inc.EnableNullIndexName = false

indexMaxLength := 767
if s.DBVersion >= 50700 {
if s.innodbLargePrefix {
indexMaxLength = 3072
}

Expand All @@ -622,7 +622,7 @@ func (s *testSessionIncSuite) TestCreateTable(c *C) {
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))

// ----------------- 索引长度审核 varchar ----------------------
if indexMaxLength == 3072 {
if s.innodbLargePrefix {
sql = "create table test_error_code_3(c1 int primary key,c2 varchar(1024),c3 int, key uq_1(c2,c3)) default charset utf8;"
s.testErrorCode(c, sql,
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))
Expand All @@ -639,12 +639,10 @@ func (s *testSessionIncSuite) TestCreateTable(c *C) {
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))

sql = "create table test_error_code_3(c1 int primary key,c2 varchar(255),c3 int, key uq_1(c2,c3)) default charset utf8;"
s.testErrorCode(c, sql,
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))
s.testErrorCode(c, sql)

sql = "create table test_error_code_3(c1 int primary key,c2 varchar(254),c3 int, key uq_1(c2,c3)) default charset utf8;"
s.testErrorCode(c, sql,
session.NewErr(session.ER_TOO_LONG_KEY, "uq_1", indexMaxLength))
s.testErrorCode(c, sql)
}

// sql = "create table test_error_code_3(c1 int,c2 text, unique uq_1(c1,c2(3068)));"
Expand Down Expand Up @@ -691,11 +689,11 @@ primary key(id)) comment 'test';`
s.testErrorCode(c, sql)

// 5.7版本新增计算列
config.GetGlobalConfig().Inc.EnableJsonType = true
if s.DBVersion >= 50700 {
sql = `CREATE TABLE t1(c1 json DEFAULT '{}' COMMENT '日志记录',
type tinyint(10) GENERATED ALWAYS AS (json_extract(operate_info, '$.type')) VIRTUAL COMMENT '操作类型')
ENGINE = InnoDB DEFAULT CHARSET = utf8 COMMENT ='xxx';`
config.GetGlobalConfig().Inc.EnableJsonType = true
s.testErrorCode(c, sql,
session.NewErr(session.ER_BLOB_CANT_HAVE_DEFAULT, "c1"))

Expand Down

0 comments on commit ce15c43

Please sign in to comment.