Skip to content

Commit

Permalink
ddl: refactor schema builder (#31172)
Browse files Browse the repository at this point in the history 
ref #30272
  • Loading branch information
@xhebox
xhebox authored Dec 31, 2021
1 parent 69b358a commit de88547
Show file tree
Hide file tree
Showing 2 changed files with 133 additions and 68 deletions.
17 changes: 13 additions & 4 deletions executor/ddl_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1486,6 +1486,10 @@ func (s *testRecoverTable) TestRenameTable(c *C) {
}()
tk := testkit.NewTestKit(c, s.store)

tk.MustExec("drop database if exists rename1")
tk.MustExec("drop database if exists rename2")
tk.MustExec("drop database if exists rename3")

tk.MustExec("create database rename1")
tk.MustExec("create database rename2")
tk.MustExec("create database rename3")
Expand Down Expand Up @@ -1563,6 +1567,11 @@ func (s *testRecoverTable) TestRenameMultiTables(c *C) {
}()
tk := testkit.NewTestKit(c, s.store)

tk.MustExec("drop database if exists rename1")
tk.MustExec("drop database if exists rename2")
tk.MustExec("drop database if exists rename3")
tk.MustExec("drop database if exists rename4")

tk.MustExec("create database rename1")
tk.MustExec("create database rename2")
tk.MustExec("create database rename3")
Expand All @@ -1577,14 +1586,14 @@ func (s *testRecoverTable) TestRenameMultiTables(c *C) {
tk.MustExec("insert rename2.t2 values ()")
tk.MustExec("drop database rename3")
tk.MustExec("insert rename4.t4 values ()")
tk.MustQuery("select * from rename2.t2").Check(testkit.Rows("1", "2"))
tk.MustQuery("select * from rename4.t4").Check(testkit.Rows("1", "2"))
tk.MustQuery("select * from rename2.t2").Check(testkit.Rows("1", "5001"))
tk.MustQuery("select * from rename4.t4").Check(testkit.Rows("1", "5001"))
// Rename a table to another table in the same database.
tk.MustExec("rename table rename2.t2 to rename2.t1, rename4.t4 to rename4.t3")
tk.MustExec("insert rename2.t1 values ()")
tk.MustQuery("select * from rename2.t1").Check(testkit.Rows("1", "2", "3"))
tk.MustQuery("select * from rename2.t1").Check(testkit.Rows("1", "5001", "10001"))
tk.MustExec("insert rename4.t3 values ()")
tk.MustQuery("select * from rename4.t3").Check(testkit.Rows("1", "2", "3"))
tk.MustQuery("select * from rename4.t3").Check(testkit.Rows("1", "5001", "10001"))
tk.MustExec("drop database rename2")
tk.MustExec("drop database rename4")

Expand Down
184 changes: 120 additions & 64 deletions infoschema/builder.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ import (
// Builder builds a new InfoSchema.
type Builder struct {
is *infoSchema
// dbInfos do not need to be copied everytime applying a diff, instead,
// they can be copied only once over the whole lifespan of Builder.
// This map will indicate which DB has been copied, so that they
// don't need to be copied again.
dirtyDB map[string]bool
// TODO: store is only used by autoid allocators
// detach allocators from storage, use passed transaction in the feature
store kv.Storage
Expand All @@ -52,6 +57,8 @@ type Builder struct {
// Return the detail updated table IDs that are produced from SchemaDiff and an error.
func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
b.is.schemaMetaVersion = diff.Version
var tblIDs []int64
var err error
switch diff.Type {
case model.ActionCreateSchema:
return nil, b.applyCreateSchema(m, diff)
Expand All @@ -67,13 +74,104 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
return b.applyDropPolicy(diff.SchemaID), nil
case model.ActionAlterPlacementPolicy:
return b.applyAlterPolicy(m, diff)
default:
switch diff.Type {
case model.ActionTruncateTablePartition, model.ActionTruncateTable:
tblIDs, err = b.applyTruncateTableOrPartition(m, diff)
case model.ActionDropTable, model.ActionDropTablePartition:
tblIDs, err = b.applyDropTableOrParition(m, diff)
case model.ActionRecoverTable:
tblIDs, err = b.applyRecoverTable(m, diff)
default:
tblIDs, err = b.applyDefaultAction(m, diff)
}
}
return tblIDs, err
}

func (b *Builder) applyTruncateTableOrPartition(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
tblIDs, err := b.applyTableUpdate(m, diff)
if err != nil {
return nil, errors.Trace(err)
}

for _, opt := range diff.AffectedOpts {
if diff.Type == model.ActionTruncateTablePartition {
// Reduce the impact on DML when executing partition DDL. eg.
// While session 1 performs the DML operation associated with partition 1,
// the TRUNCATE operation of session 2 on partition 2 does not cause the operation of session 1 to fail.
tblIDs = append(tblIDs, opt.OldTableID)
}
b.applyPlacementDelete(placement.GroupID(opt.OldTableID))
err := b.applyPlacementUpdate(placement.GroupID(opt.TableID))
if err != nil {
return nil, errors.Trace(err)
}
}
return tblIDs, nil
}

func (b *Builder) applyDropTableOrParition(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
tblIDs, err := b.applyTableUpdate(m, diff)
if err != nil {
return nil, errors.Trace(err)
}

for _, opt := range diff.AffectedOpts {
b.applyPlacementDelete(placement.GroupID(opt.OldTableID))
}
return tblIDs, nil
}

func (b *Builder) applyRecoverTable(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
tblIDs, err := b.applyTableUpdate(m, diff)
if err != nil {
return nil, errors.Trace(err)
}

for _, opt := range diff.AffectedOpts {
err := b.applyPlacementUpdate(placement.GroupID(opt.TableID))
if err != nil {
return nil, errors.Trace(err)
}
}
return tblIDs, nil
}

func (b *Builder) applyDefaultAction(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
tblIDs, err := b.applyTableUpdate(m, diff)
if err != nil {
return nil, errors.Trace(err)
}

for _, opt := range diff.AffectedOpts {
var err error
affectedDiff := &model.SchemaDiff{
Version: diff.Version,
Type: diff.Type,
SchemaID: opt.SchemaID,
TableID: opt.TableID,
OldSchemaID: opt.OldSchemaID,
OldTableID: opt.OldTableID,
}
affectedIDs, err := b.ApplyDiff(m, affectedDiff)
if err != nil {
return nil, errors.Trace(err)
}
tblIDs = append(tblIDs, affectedIDs...)
}

return tblIDs, nil
}

func (b *Builder) applyTableUpdate(m *meta.Meta, diff *model.SchemaDiff) ([]int64, error) {
roDBInfo, ok := b.is.SchemaByID(diff.SchemaID)
if !ok {
return nil, ErrDatabaseNotExists.GenWithStackByArgs(
fmt.Sprintf("(Schema ID %d)", diff.SchemaID),
)
}
dbInfo := b.getSchemaAndCopyIfNecessary(roDBInfo.Name.L)
var oldTableID, newTableID int64
switch diff.Type {
case model.ActionCreateTable, model.ActionCreateSequence, model.ActionRecoverTable:
Expand Down Expand Up @@ -109,14 +207,13 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
return nil, errors.Trace(err)
}
}
dbInfo := b.copySchemaTables(roDBInfo.Name.L)
b.copySortedTables(oldTableID, newTableID)

tblIDs := make([]int64, 0, 2)
// We try to reuse the old allocator, so the cached auto ID can be reused.
var allocs autoid.Allocators
if tableIDIsValid(oldTableID) {
if oldTableID == newTableID && diff.Type != model.ActionRenameTable &&
if oldTableID == newTableID && (diff.Type != model.ActionRenameTable && diff.Type != model.ActionRenameTables) &&
diff.Type != model.ActionExchangeTablePartition &&
// For repairing table in TiDB cluster, given 2 normal node and 1 repair node.
// For normal node's information schema, repaired table is existed.
Expand All @@ -137,7 +234,7 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
fmt.Sprintf("(Schema ID %d)", diff.OldSchemaID),
)
}
oldDBInfo := b.copySchemaTables(oldRoDBInfo.Name.L)
oldDBInfo := b.getSchemaAndCopyIfNecessary(oldRoDBInfo.Name.L)
tmpIDs = b.applyDropTable(oldDBInfo, oldTableID, tmpIDs)
} else {
tmpIDs = b.applyDropTable(dbInfo, oldTableID, tmpIDs)
Expand All @@ -156,53 +253,6 @@ func (b *Builder) ApplyDiff(m *meta.Meta, diff *model.SchemaDiff) ([]int64, erro
return nil, errors.Trace(err)
}
}
if diff.AffectedOpts != nil {
for _, opt := range diff.AffectedOpts {
switch diff.Type {
case model.ActionTruncateTablePartition:
// Reduce the impact on DML when executing partition DDL. eg.
// While session 1 performs the DML operation associated with partition 1,
// the TRUNCATE operation of session 2 on partition 2 does not cause the operation of session 1 to fail.
tblIDs = append(tblIDs, opt.OldTableID)
b.applyPlacementDelete(placement.GroupID(opt.OldTableID))
err := b.applyPlacementUpdate(placement.GroupID(opt.TableID))
if err != nil {
return nil, errors.Trace(err)
}
continue
case model.ActionDropTable, model.ActionDropTablePartition:
b.applyPlacementDelete(placement.GroupID(opt.OldTableID))
continue
case model.ActionTruncateTable:
b.applyPlacementDelete(placement.GroupID(opt.OldTableID))
err := b.applyPlacementUpdate(placement.GroupID(opt.TableID))
if err != nil {
return nil, errors.Trace(err)
}
continue
case model.ActionRecoverTable:
err := b.applyPlacementUpdate(placement.GroupID(opt.TableID))
if err != nil {
return nil, errors.Trace(err)
}
continue
}
var err error
affectedDiff := &model.SchemaDiff{
Version: diff.Version,
Type: diff.Type,
SchemaID: opt.SchemaID,
TableID: opt.TableID,
OldSchemaID: opt.OldSchemaID,
OldTableID: opt.OldTableID,
}
affectedIDs, err := b.ApplyDiff(m, affectedDiff)
if err != nil {
return nil, errors.Trace(err)
}
tblIDs = append(tblIDs, affectedIDs...)
}
}
return tblIDs, nil
}

Expand Down Expand Up @@ -306,7 +356,7 @@ func (b *Builder) applyModifySchemaCharsetAndCollate(m *meta.Meta, diff *model.S
fmt.Sprintf("(Schema ID %d)", diff.SchemaID),
)
}
newDbInfo := b.copySchemaTables(di.Name.L)
newDbInfo := b.getSchemaAndCopyIfNecessary(di.Name.L)
newDbInfo.Charset = di.Charset
newDbInfo.Collate = di.Collate
return nil
Expand All @@ -323,7 +373,7 @@ func (b *Builder) applyModifySchemaDefaultPlacement(m *meta.Meta, diff *model.Sc
fmt.Sprintf("(Schema ID %d)", diff.SchemaID),
)
}
newDbInfo := b.copySchemaTables(di.Name.L)
newDbInfo := b.getSchemaAndCopyIfNecessary(di.Name.L)
newDbInfo.PlacementPolicyRef = di.PlacementPolicyRef
newDbInfo.DirectPlacementOpts = di.DirectPlacementOpts
return nil
Expand Down Expand Up @@ -577,20 +627,25 @@ func (b *Builder) copyPoliciesMap(oldIS *infoSchema) {
}
}

// copySchemaTables creates a new schemaTables instance when a table in the database has changed.
// getSchemaAndCopyIfNecessary creates a new schemaTables instance when a table in the database has changed.
// It also does modifications on the new one because old schemaTables must be read-only.
// Note: please make sure the dbName is in lowercase.
func (b *Builder) copySchemaTables(dbName string) *model.DBInfo {
oldSchemaTables := b.is.schemaMap[dbName]
newSchemaTables := &schemaTables{
dbInfo: oldSchemaTables.dbInfo.Copy(),
tables: make(map[string]table.Table, len(oldSchemaTables.tables)),
}
for k, v := range oldSchemaTables.tables {
newSchemaTables.tables[k] = v
// And it will only copy the changed database once in the lifespan of the Builder.
// NOTE: please make sure the dbName is in lowercase.
func (b *Builder) getSchemaAndCopyIfNecessary(dbName string) *model.DBInfo {
if !b.dirtyDB[dbName] {
b.dirtyDB[dbName] = true
oldSchemaTables := b.is.schemaMap[dbName]
newSchemaTables := &schemaTables{
dbInfo: oldSchemaTables.dbInfo.Copy(),
tables: make(map[string]table.Table, len(oldSchemaTables.tables)),
}
for k, v := range oldSchemaTables.tables {
newSchemaTables.tables[k] = v
}
b.is.schemaMap[dbName] = newSchemaTables
return newSchemaTables.dbInfo
}
b.is.schemaMap[dbName] = newSchemaTables
return newSchemaTables.dbInfo
return b.is.schemaMap[dbName].dbInfo
}

// InitWithDBInfos initializes an empty new InfoSchema with a slice of DBInfo, all placement rules, and schema version.
Expand Down Expand Up @@ -692,6 +747,7 @@ func NewBuilder(store kv.Storage, renewCh chan func(), factory func() (pools.Res
ruleBundleMap: map[string]*placement.Bundle{},
sortedTablesBuckets: make([]sortedTables, bucketCount),
},
dirtyDB: make(map[string]bool),
renewLeaseCh: renewCh,
factory: factory,
}
Expand Down

0 comments on commit de88547

Please sign in to comment.