Skip to content

Commit

Permalink
ddl, executor: fix rename table compatibility (pingcap#8709)
Browse files Browse the repository at this point in the history
  • Loading branch information
@zimulala
zimulala committed Dec 25, 2018
1 parent accf34c commit 4546161
Show file tree
Hide file tree
Showing 7 changed files with 262 additions and 244 deletions.
2 changes: 1 addition & 1 deletion ddl/ddl.go
View file
Expand Up @@ -166,7 +166,7 @@ type DDL interface {
GetInformationSchema() infoschema.InfoSchema
AlterTable(ctx sessionctx.Context, tableIdent ast.Ident, spec []*ast.AlterTableSpec) error
TruncateTable(ctx sessionctx.Context, tableIdent ast.Ident) error
RenameTable(ctx sessionctx.Context, oldTableIdent, newTableIdent ast.Ident) error
RenameTable(ctx sessionctx.Context, oldTableIdent, newTableIdent ast.Ident, isAlterTable bool) error
// SetLease will reset the lease time for online DDL change,
// it's a very dangerous function and you must guarantee that all servers have the same lease time.
SetLease(ctx context.Context, lease time.Duration)
Expand Down
9 changes: 7 additions & 2 deletions ddl/ddl_api.go
View file
Expand Up @@ -994,7 +994,8 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
err = d.AlterColumn(ctx, ident, spec)
case ast.AlterTableRenameTable:
newIdent := ast.Ident{Schema: spec.NewTable.Schema, Name: spec.NewTable.Name}
err = d.RenameTable(ctx, ident, newIdent)
isAlterTable := true
err = d.RenameTable(ctx, ident, newIdent, isAlterTable)
case ast.AlterTableDropPrimaryKey:
err = ErrUnsupportedModifyPrimaryKey.GenByArgs("drop")
case ast.AlterTableOption:
Expand Down Expand Up @@ -1619,7 +1620,7 @@ func (d *ddl) TruncateTable(ctx sessionctx.Context, ti ast.Ident) error {
return errors.Trace(err)
}

func (d *ddl) RenameTable(ctx sessionctx.Context, oldIdent, newIdent ast.Ident) error {
func (d *ddl) RenameTable(ctx sessionctx.Context, oldIdent, newIdent ast.Ident, isAlterTable bool) error {
is := d.GetInformationSchema()
oldSchema, ok := is.SchemaByName(oldIdent.Schema)
if !ok {
Expand All @@ -1629,6 +1630,10 @@ func (d *ddl) RenameTable(ctx sessionctx.Context, oldIdent, newIdent ast.Ident)
if err != nil {
return errFileNotFound.GenByArgs(oldIdent.Schema, oldIdent.Name)
}
if isAlterTable && newIdent.Schema.L == oldIdent.Schema.L && newIdent.Name.L == oldIdent.Name.L {
// oldIdent is equal to newIdent, do nothing
return nil
}
newSchema, ok := is.SchemaByName(newIdent.Schema)
if !ok {
return errErrorOnRename.GenByArgs(oldIdent.Schema, oldIdent.Name, newIdent.Schema, newIdent.Name)
Expand Down
22 changes: 17 additions & 5 deletions ddl/ddl_db_test.go
View file
Expand Up @@ -1590,14 +1590,16 @@ func (s *testDBSuite) TestTruncateTable(c *C) {
}

func (s *testDBSuite) TestRenameTable(c *C) {
s.testRenameTable(c, "rename table %s to %s")
isAlterTable := false
s.testRenameTable(c, "rename table %s to %s", isAlterTable)
}

func (s *testDBSuite) TestAlterTableRenameTable(c *C) {
s.testRenameTable(c, "alter table %s rename to %s")
isAlterTable := true
s.testRenameTable(c, "alter table %s rename to %s", isAlterTable)
}

func (s *testDBSuite) testRenameTable(c *C, sql string) {
func (s *testDBSuite) testRenameTable(c *C, sql string, isAlterTable bool) {
s.tk = testkit.NewTestKit(c, s.store)
s.tk.MustExec("use test")
// for different databases
Expand Down Expand Up @@ -1640,8 +1642,18 @@ func (s *testDBSuite) testRenameTable(c *C, sql string) {
s.testErrorCode(c, failSQL, tmysql.ErrFileNotFound)
failSQL = fmt.Sprintf(sql, "test1.t2", "test_not_exist.t")
s.testErrorCode(c, failSQL, tmysql.ErrErrorOnRename)
failSQL = fmt.Sprintf(sql, "test1.t2", "test1.t2")
s.testErrorCode(c, failSQL, tmysql.ErrTableExists)

// for the same table name
s.tk.MustExec("use test1")
s.tk.MustExec("create table if not exists t (c1 int, c2 int)")
s.tk.MustExec("create table if not exists t1 (c1 int, c2 int)")
if isAlterTable {
s.tk.MustExec(fmt.Sprintf(sql, "test1.t", "t"))
s.tk.MustExec(fmt.Sprintf(sql, "test1.t1", "test1.T1"))
} else {
s.testErrorCode(c, fmt.Sprintf(sql, "test1.t", "t"), tmysql.ErrTableExists)
s.testErrorCode(c, fmt.Sprintf(sql, "test1.t1", "test1.T1"), tmysql.ErrTableExists)
}

s.tk.MustExec("drop database test1")
}
Expand Down
3 changes: 2 additions & 1 deletion executor/ddl.go
View file
Expand Up @@ -93,7 +93,8 @@ func (e *DDLExec) executeRenameTable(s *ast.RenameTableStmt) error {
}
oldIdent := ast.Ident{Schema: s.OldTable.Schema, Name: s.OldTable.Name}
newIdent := ast.Ident{Schema: s.NewTable.Schema, Name: s.NewTable.Name}
err := domain.GetDomain(e.ctx).DDL().RenameTable(e.ctx, oldIdent, newIdent)
isAlterTable := false
err := domain.GetDomain(e.ctx).DDL().RenameTable(e.ctx, oldIdent, newIdent, isAlterTable)
return errors.Trace(err)
}

Expand Down
24 changes: 12 additions & 12 deletions executor/show.go
View file
Expand Up @@ -345,19 +345,19 @@ func (e *ShowExec) fetchShowIndex() error {
subPart = col.Length
}
e.appendRow([]interface{}{
tb.Meta().Name.O, // Table
nonUniq, // Non_unique
idx.Meta().Name.O, // Key_name
i + 1, // Seq_in_index
col.Name.O, // Column_name
"A", // Collation
0, // Cardinality
subPart, // Sub_part
nil, // Packed
"YES", // Null
tb.Meta().Name.O, // Table
nonUniq, // Non_unique
idx.Meta().Name.O, // Key_name
i + 1, // Seq_in_index
col.Name.O, // Column_name
"A", // Collation
0, // Cardinality
subPart, // Sub_part
nil, // Packed
"YES", // Null
idx.Meta().Tp.String(), // Index_type
"", // Comment
idx.Meta().Comment, // Index_comment
"", // Comment
idx.Meta().Comment, // Index_comment
})
}
}
Expand Down
54 changes: 27 additions & 27 deletions infoschema/tables.go
View file
Expand Up @@ -775,18 +775,18 @@ func dataForColumnsInTable(schema *model.DBInfo, tbl *model.TableInfo) [][]types
columnDefault, // COLUMN_DEFAULT
columnDesc.Null, // IS_NULLABLE
types.TypeToStr(col.Tp, col.Charset), // DATA_TYPE
colLen, // CHARACTER_MAXIMUM_LENGTH
colLen, // CHARACTER_OCTET_LENGTH
decimal, // NUMERIC_PRECISION
0, // NUMERIC_SCALE
0, // DATETIME_PRECISION
col.Charset, // CHARACTER_SET_NAME
col.Collate, // COLLATION_NAME
columnType, // COLUMN_TYPE
columnDesc.Key, // COLUMN_KEY
columnDesc.Extra, // EXTRA
"select,insert,update,references", // PRIVILEGES
columnDesc.Comment, // COLUMN_COMMENT
colLen, // CHARACTER_MAXIMUM_LENGTH
colLen, // CHARACTER_OCTET_LENGTH
decimal, // NUMERIC_PRECISION
0, // NUMERIC_SCALE
0, // DATETIME_PRECISION
col.Charset, // CHARACTER_SET_NAME
col.Collate, // COLLATION_NAME
columnType, // COLUMN_TYPE
columnDesc.Key, // COLUMN_KEY
columnDesc.Extra, // EXTRA
"select,insert,update,references", // PRIVILEGES
columnDesc.Comment, // COLUMN_COMMENT
)
// In mysql, 'character_set_name' and 'collation_name' are setted to null when column type is non-varchar or non-blob in information_schema.
if col.Tp != mysql.TypeVarchar && col.Tp != mysql.TypeBlob {
Expand Down Expand Up @@ -930,24 +930,24 @@ func dataForTableConstraints(schemas []*model.DBInfo) [][]types.Datum {
func dataForPseudoProfiling() [][]types.Datum {
var rows [][]types.Datum
row := types.MakeDatums(
0, // QUERY_ID
0, // SEQ
"", // STATE
0, // QUERY_ID
0, // SEQ
"", // STATE
types.NewDecFromInt(0), // DURATION
types.NewDecFromInt(0), // CPU_USER
types.NewDecFromInt(0), // CPU_SYSTEM
0, // CONTEXT_VOLUNTARY
0, // CONTEXT_INVOLUNTARY
0, // BLOCK_OPS_IN
0, // BLOCK_OPS_OUT
0, // MESSAGES_SENT
0, // MESSAGES_RECEIVED
0, // PAGE_FAULTS_MAJOR
0, // PAGE_FAULTS_MINOR
0, // SWAPS
0, // SOURCE_FUNCTION
0, // SOURCE_FILE
0, // SOURCE_LINE
0, // CONTEXT_VOLUNTARY
0, // CONTEXT_INVOLUNTARY
0, // BLOCK_OPS_IN
0, // BLOCK_OPS_OUT
0, // MESSAGES_SENT
0, // MESSAGES_RECEIVED
0, // PAGE_FAULTS_MAJOR
0, // PAGE_FAULTS_MINOR
0, // SWAPS
0, // SOURCE_FUNCTION
0, // SOURCE_FILE
0, // SOURCE_LINE
)
rows = append(rows, row)
return rows
Expand Down

0 comments on commit 4546161

Please sign in to comment.