Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

schematracker: treat TableInfo as immutable #39103

Merged
merged 3 commits into from
Nov 14, 2022
Merged

schematracker: treat TableInfo as immutable #39103

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8ef53bd
schematracker: treat TableInfo as immutable
lance6716 Nov 13, 2022
e896b64
address comment
lance6716 Nov 14, 2022
84c09aa
Merge branch 'master' into immutable-tableinfo
ti-chi-bot Nov 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 118 additions & 14 deletions ddl/schematracker/dm_tracker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,9 @@ var _ ddl.DDL = SchemaTracker{}

// SchemaTracker is used to track schema changes by DM. It implements DDL interface and by applying DDL, it updates the
// table structure to keep tracked with upstream changes.
// It embeds an InfoStore which stores DBInfo and TableInfo. The DBInfo and TableInfo can be treated as immutable, so
// after reading them by SchemaByName or TableByName, later modifications made by SchemaTracker will not change them.
// SchemaTracker is not thread-safe.
type SchemaTracker struct {
*InfoStore
}
Expand Down Expand Up @@ -108,16 +111,22 @@ func (d SchemaTracker) CreateSchemaWithInfo(ctx sessionctx.Context, dbInfo *mode
}

// AlterSchema implements the DDL interface.
func (d SchemaTracker) AlterSchema(ctx sessionctx.Context, stmt *ast.AlterDatabaseStmt) error {
func (d SchemaTracker) AlterSchema(ctx sessionctx.Context, stmt *ast.AlterDatabaseStmt) (err error) {
dbInfo := d.SchemaByName(stmt.Name)
if dbInfo == nil {
return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(stmt.Name.O)
}

dbInfo = dbInfo.Clone()
defer func() {
if err == nil {
d.PutSchema(dbInfo)
}
}()

// Resolve target charset and collation from options.
var (
toCharset, toCollate string
err error
)

for _, val := range stmt.Options {
Expand Down Expand Up @@ -173,9 +182,15 @@ func (d SchemaTracker) CreateTable(ctx sessionctx.Context, s *ast.CreateTableStm
return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(ident.Schema)
}
// suppress ErrTooLongKey
strictSQLModeBackup := ctx.GetSessionVars().StrictSQLMode
ctx.GetSessionVars().StrictSQLMode = false
// support drop PK
enableClusteredIndexBackup := ctx.GetSessionVars().EnableClusteredIndex
ctx.GetSessionVars().EnableClusteredIndex = variable.ClusteredIndexDefModeOff
defer func() {
ctx.GetSessionVars().StrictSQLMode = strictSQLModeBackup
ctx.GetSessionVars().EnableClusteredIndex = enableClusteredIndexBackup
}()

var (
referTbl *model.TableInfo
Expand Down Expand Up @@ -363,12 +378,20 @@ func (d SchemaTracker) createIndex(
indexPartSpecifications []*ast.IndexPartSpecification,
indexOption *ast.IndexOption,
ifNotExists bool,
) error {
) (err error) {
unique := keyType == ast.IndexKeyTypeUnique
tblInfo, err := d.TableByName(ti.Schema, ti.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ti.Schema, tblInfo)
}
}()

t := tables.MockTableFromMeta(tblInfo)

// Deal with anonymous index.
Expand Down Expand Up @@ -432,12 +455,19 @@ func (d SchemaTracker) DropIndex(ctx sessionctx.Context, stmt *ast.DropIndexStmt
}

// dropIndex is shared by DropIndex and AlterTable.
func (d SchemaTracker) dropIndex(ctx sessionctx.Context, ti ast.Ident, indexName model.CIStr, ifExists bool) error {
func (d SchemaTracker) dropIndex(ctx sessionctx.Context, ti ast.Ident, indexName model.CIStr, ifExists bool) (err error) {
tblInfo, err := d.TableByName(ti.Schema, ti.Name)
if err != nil {
return infoschema.ErrTableNotExists.GenWithStackByArgs(ti.Schema, ti.Name)
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ti.Schema, tblInfo)
}
}()

indexInfo := tblInfo.FindIndexByName(indexName.L)
if indexInfo == nil {
if ifExists {
Expand All @@ -464,7 +494,7 @@ func (d SchemaTracker) dropIndex(ctx sessionctx.Context, ti ast.Ident, indexName
}

// addColumn is used by AlterTable.
func (d SchemaTracker) addColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) addColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
specNewColumn := spec.NewColumns[0]
schema := d.SchemaByName(ti.Schema)
if schema == nil {
Expand All @@ -474,6 +504,14 @@ func (d SchemaTracker) addColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ti.Schema, tblInfo)
}
}()

t := tables.MockTableFromMeta(tblInfo)
colName := specNewColumn.Name.Name.O

Expand Down Expand Up @@ -504,12 +542,19 @@ func (d SchemaTracker) addColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast
}

// dropColumn is used by AlterTable.
func (d *SchemaTracker) dropColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTableSpec) error {
func (d *SchemaTracker) dropColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
tblInfo, err := d.TableByName(ti.Schema, ti.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ti.Schema, tblInfo)
}
}()

colName := spec.OldColumnName.Name
colInfo := tblInfo.FindPublicColumnByName(colName.L)
if colInfo == nil {
Expand Down Expand Up @@ -546,14 +591,22 @@ func (d *SchemaTracker) dropColumn(ctx sessionctx.Context, ti ast.Ident, spec *a
}

// renameColumn is used by AlterTable.
func (d SchemaTracker) renameColumn(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) renameColumn(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
oldColName := spec.OldColumnName.Name
newColName := spec.NewColumnName.Name

tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

tbl := tables.MockTableFromMeta(tblInfo)

oldCol := table.FindCol(tbl.VisibleCols(), oldColName.L)
Expand Down Expand Up @@ -595,12 +648,20 @@ func (d SchemaTracker) renameColumn(ctx sessionctx.Context, ident ast.Ident, spe
}

// alterColumn is used by AlterTable.
func (d SchemaTracker) alterColumn(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) alterColumn(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
specNewColumn := spec.NewColumns[0]
tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

t := tables.MockTableFromMeta(tblInfo)

colName := specNewColumn.Name.Name
Expand Down Expand Up @@ -664,11 +725,19 @@ func (d SchemaTracker) handleModifyColumn(
ident ast.Ident,
originalColName model.CIStr,
spec *ast.AlterTableSpec,
) error {
) (err error) {
tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

schema := d.SchemaByName(ident.Schema)
t := tables.MockTableFromMeta(tblInfo)
job, err := ddl.GetModifiableColumnJob(ctx, sctx, nil, ident, originalColName, schema, t, spec)
Expand Down Expand Up @@ -714,11 +783,19 @@ func (d SchemaTracker) handleModifyColumn(
}

// renameIndex is used by AlterTable.
func (d SchemaTracker) renameIndex(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) renameIndex(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return err
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

duplicate, err := ddl.ValidateRenameIndex(spec.FromKey, spec.ToKey, tblInfo)
if duplicate {
return nil
Expand All @@ -732,12 +809,19 @@ func (d SchemaTracker) renameIndex(ctx sessionctx.Context, ident ast.Ident, spec
}

// addTablePartitions is used by AlterTable.
func (d SchemaTracker) addTablePartitions(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) addTablePartitions(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return errors.Trace(err)
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

pi := tblInfo.GetPartitionInfo()
if pi == nil {
return errors.Trace(dbterror.ErrPartitionMgmtOnNonpartitioned)
Expand All @@ -752,12 +836,19 @@ func (d SchemaTracker) addTablePartitions(ctx sessionctx.Context, ident ast.Iden
}

// dropTablePartitions is used by AlterTable.
func (d SchemaTracker) dropTablePartition(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) error {
func (d SchemaTracker) dropTablePartitions(ctx sessionctx.Context, ident ast.Ident, spec *ast.AlterTableSpec) (err error) {
tblInfo, err := d.TableByName(ident.Schema, ident.Name)
if err != nil {
return errors.Trace(err)
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ident.Schema, tblInfo)
}
}()

pi := tblInfo.GetPartitionInfo()
if pi == nil {
return errors.Trace(dbterror.ErrPartitionMgmtOnNonpartitioned)
Expand Down Expand Up @@ -799,12 +890,19 @@ func (d SchemaTracker) createPrimaryKey(
indexName model.CIStr,
indexPartSpecifications []*ast.IndexPartSpecification,
indexOption *ast.IndexOption,
) error {
) (err error) {
tblInfo, err := d.TableByName(ti.Schema, ti.Name)
if err != nil {
return errors.Trace(err)
}

tblInfo = tblInfo.Clone()
defer func() {
if err == nil {
_ = d.PutTable(ti.Schema, tblInfo)
}
}()

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this part code has many same parts in the file.

indexName = model.NewCIStr(mysql.PrimaryKeyName)
if indexInfo := tblInfo.FindIndexByName(indexName.L); indexInfo != nil ||
// If the table's PKIsHandle is true, it also means that this table has a primary key.
Expand Down Expand Up @@ -888,7 +986,7 @@ func (d SchemaTracker) AlterTable(ctx context.Context, sctx sessionctx.Context,
case ast.AlterTableRenameIndex:
err = d.renameIndex(sctx, ident, spec)
case ast.AlterTableDropPartition:
err = d.dropTablePartition(sctx, ident, spec)
err = d.dropTablePartitions(sctx, ident, spec)
case ast.AlterTableAddConstraint:
constr := spec.Constraint
switch spec.Constraint.Tp {
Expand Down Expand Up @@ -925,7 +1023,9 @@ func (d SchemaTracker) AlterTable(ctx context.Context, sctx sessionctx.Context,
case ast.TableOptionAutoIdCache:
case ast.TableOptionAutoRandomBase:
case ast.TableOptionComment:
tblInfo = tblInfo.Clone()
tblInfo.Comment = opt.StrValue
_ = d.PutTable(ident.Schema, tblInfo)
case ast.TableOptionCharset, ast.TableOptionCollate:
// getCharsetAndCollateInTableOption will get the last charset and collate in the options,
// so it should be handled only once.
Expand All @@ -939,6 +1039,7 @@ func (d SchemaTracker) AlterTable(ctx context.Context, sctx sessionctx.Context,
}
needsOverwriteCols := ddl.NeedToOverwriteColCharset(spec.Options)

tblInfo = tblInfo.Clone()
if toCharset != "" {
tblInfo.Charset = toCharset
}
Expand All @@ -957,6 +1058,7 @@ func (d SchemaTracker) AlterTable(ctx context.Context, sctx sessionctx.Context,
}
}
}
_ = d.PutTable(ident.Schema, tblInfo)

handledCharsetOrCollate = true
case ast.TableOptionPlacementPolicy:
Expand All @@ -970,11 +1072,13 @@ func (d SchemaTracker) AlterTable(ctx context.Context, sctx sessionctx.Context,
}
}
case ast.AlterTableIndexInvisible:
tblInfo = tblInfo.Clone()
idx := tblInfo.FindIndexByName(spec.IndexName.L)
if idx == nil {
return errors.Trace(infoschema.ErrKeyNotExists.GenWithStackByArgs(spec.IndexName.O, ident.Name))
}
idx.Invisible = spec.Visibility == ast.IndexVisibilityInvisible
_ = d.PutTable(ident.Schema, tblInfo)
case ast.AlterTablePartitionOptions,
ast.AlterTableDropForeignKey,
ast.AlterTableCoalescePartitions,
Expand Down
Loading