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

ddl: correct the job.SchemaState in DROP cases #34235

Merged
merged 8 commits into from
Apr 28, 2022
Merged

ddl: correct the job.SchemaState in DROP cases #34235

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d4b7f76
ddl: correct the job.SchemaState in DROP cases
tangenta Apr 26, 2022
4ab6e53
fix integration tests
tangenta Apr 26, 2022
f04d7b9
correct job.SchemaState for drop partition
tangenta Apr 28, 2022
0691d40
revert unnecessary change
tangenta Apr 28, 2022
99a6ee4
remove unnecessary change
tangenta Apr 28, 2022
d539d8d
address comment
tangenta Apr 28, 2022
c419f8f
Merge branch 'master' into fix-test-cancel
ti-chi-bot Apr 28, 2022
eee54ce
fix integration test TestDropAndTruncatePartition
tangenta Apr 28, 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
43 changes: 20 additions & 23 deletions ddl/cancel_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,7 @@ import (
"github.com/pingcap/tidb/errno"
"github.com/pingcap/tidb/parser/model"
"github.com/pingcap/tidb/testkit"
"github.com/pingcap/tidb/util/logutil"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
)

type testCancelJob struct {
Expand All @@ -49,8 +47,7 @@ var allTestCase = []testCancelJob{
{"alter table t add primary key idx_pc2 (c2)", true, model.StateWriteReorganization, true, true, nil},
{"alter table t add primary key idx_pc2 (c2)", false, model.StatePublic, false, true, nil},
// Drop primary key
// TODO: fix schema state.
{"alter table t drop primary key", true, model.StateNone, true, false, nil},
{"alter table t drop primary key", true, model.StatePublic, true, false, nil},
{"alter table t drop primary key", false, model.StateWriteOnly, true, false, nil},
{"alter table t drop primary key", false, model.StateWriteOnly, true, false, []string{"alter table t add primary key idx_pc2 (c2)"}},
{"alter table t drop primary key", false, model.StateDeleteOnly, true, false, []string{"alter table t add primary key idx_pc2 (c2)"}},
Expand All @@ -75,36 +72,32 @@ var allTestCase = []testCancelJob{
{"create table test_create_table(a int)", true, model.StateNone, true, false, nil},
{"create table test_create_table(a int)", false, model.StatePublic, false, true, nil},
// Drop table.
// TODO: fix schema state.
{"drop table test_create_table", true, model.StateNone, true, false, nil},
{"drop table test_create_table", true, model.StatePublic, true, false, nil},
{"drop table test_create_table", false, model.StateWriteOnly, true, true, []string{"create table if not exists test_create_table(a int)"}},
{"drop table test_create_table", false, model.StateDeleteOnly, true, true, []string{"create table if not exists test_create_table(a int)"}},
{"drop table test_create_table", false, model.StatePublic, false, true, []string{"create table if not exists test_create_table(a int)"}},
{"drop table test_create_table", false, model.StateNone, false, true, []string{"create table if not exists test_create_table(a int)"}},
// Create schema.
{"create database test_create_db", true, model.StateNone, true, false, nil},
{"create database test_create_db", false, model.StatePublic, false, true, nil},
// Drop schema.
// TODO: fix schema state.
{"drop database test_create_db", true, model.StateNone, true, false, nil},
{"drop database test_create_db", true, model.StatePublic, true, false, nil},
{"drop database test_create_db", false, model.StateWriteOnly, true, true, []string{"create database if not exists test_create_db"}},
{"drop database test_create_db", false, model.StateDeleteOnly, true, true, []string{"create database if not exists test_create_db"}},
{"drop database test_create_db", false, model.StatePublic, false, true, []string{"create database if not exists test_create_db"}},
{"drop database test_create_db", false, model.StateNone, false, true, []string{"create database if not exists test_create_db"}},
// Drop column.
// TODO: fix schema state.
{"alter table t drop column c3", true, model.StateNone, true, false, nil},
{"alter table t drop column c3", true, model.StatePublic, true, false, nil},
{"alter table t drop column c3", false, model.StateDeleteOnly, true, false, nil},
{"alter table t drop column c3", false, model.StateDeleteOnly, false, true, []string{"alter table t add column c3 bigint"}},
{"alter table t drop column c3", false, model.StateWriteOnly, true, true, []string{"alter table t add column c3 bigint"}},
{"alter table t drop column c3", false, model.StateDeleteReorganization, true, true, []string{"alter table t add column c3 bigint"}},
{"alter table t drop column c3", false, model.StatePublic, false, true, []string{"alter table t add column c3 bigint"}},
{"alter table t drop column c3", false, model.StateNone, false, true, []string{"alter table t add column c3 bigint"}},
// Drop column with index.
// TODO: fix schema state.
{"alter table t drop column c3", true, model.StateNone, true, false, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", true, model.StatePublic, true, false, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", false, model.StateDeleteOnly, true, false, nil},
{"alter table t drop column c3", false, model.StateDeleteOnly, false, true, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", false, model.StateWriteOnly, true, true, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", false, model.StateDeleteReorganization, true, true, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", false, model.StatePublic, false, true, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
{"alter table t drop column c3", false, model.StateNone, false, true, []string{"alter table t add column c3 bigint", "alter table t add index idx_c3(c3)"}},
// rebase auto ID.
{"alter table t_rebase auto_increment = 6000", true, model.StateNone, true, false, []string{"create table t_rebase (c1 bigint auto_increment primary key, c2 bigint);"}},
{"alter table t_rebase auto_increment = 9000", false, model.StatePublic, false, true, nil},
Expand All @@ -128,9 +121,8 @@ var allTestCase = []testCancelJob{
{"alter table t add constraint fk foreign key a(c1) references t_ref(c1)", true, model.StateNone, true, false, []string{"create table t_ref (c1 int, c2 int, c3 int, c11 tinyint);"}},
{"alter table t add constraint fk foreign key a(c1) references t_ref(c1)", false, model.StatePublic, false, true, nil},
// Drop foreign key.
// TODO: fix schema state.
{"alter table t drop foreign key fk", true, model.StateNone, true, false, nil},
{"alter table t drop foreign key fk", false, model.StatePublic, false, true, nil},
{"alter table t drop foreign key fk", true, model.StatePublic, true, false, nil},
{"alter table t drop foreign key fk", false, model.StateNone, false, true, nil},
// Rename table.
{"rename table t_rename1 to t_rename11", true, model.StateNone, true, false, []string{"create table t_rename1 (c1 bigint , c2 bigint);", "create table t_rename2 (c1 bigint , c2 bigint);"}},
{"rename table t_rename1 to t_rename11", false, model.StatePublic, false, true, nil},
Expand Down Expand Up @@ -242,6 +234,7 @@ func TestCancel(t *testing.T) {
hook := &ddl.TestDDLCallback{Do: dom}
i := 0
cancel := false
cancelResult := false
cancelWhenReorgNotStart := false

hookFunc := func(job *model.Job) {
Expand All @@ -250,7 +243,7 @@ func TestCancel(t *testing.T) {
return
}
rs := tkCancel.MustQuery(fmt.Sprintf("admin cancel ddl jobs %d", job.ID))
require.Equal(t, allTestCase[i].ok, cancelSuccess(rs))
cancelResult = cancelSuccess(rs)
cancel = true
}
}
Expand All @@ -270,6 +263,7 @@ func TestCancel(t *testing.T) {

for j, tc := range allTestCase {
i = j
msg := fmt.Sprintf("sql: %s, state: %s", tc.sql, tc.cancelState)
if tc.onJobBefore {
restHook(hook)
for _, prepareSQL := range tc.prepareSQL {
Expand All @@ -279,28 +273,31 @@ func TestCancel(t *testing.T) {
cancel = false
cancelWhenReorgNotStart = true
registHook(hook, true)
logutil.BgLogger().Info("test case", zap.Int("", i))
if tc.ok {
tk.MustGetErrCode(tc.sql, errno.ErrCancelledDDLJob)
} else {
tk.MustExec(tc.sql)
}
if cancel {
require.Equal(t, tc.ok, cancelResult, msg)
}
}
if tc.onJobUpdate {
restHook(hook)
for _, prepareSQL := range tc.prepareSQL {
tk.MustExec(prepareSQL)
}

cancel = false
cancelWhenReorgNotStart = false
registHook(hook, false)
logutil.BgLogger().Info("test case", zap.Int("", i))
if tc.ok {
tk.MustGetErrCode(tc.sql, errno.ErrCancelledDDLJob)
} else {
tk.MustExec(tc.sql)
}
if cancel {
require.Equal(t, tc.ok, cancelResult, msg)
}
}
}
}
4 changes: 1 addition & 3 deletions ddl/column.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -556,7 +556,6 @@ func onDropColumn(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
if err != nil {
return ver, errors.Trace(err)
}
job.SchemaState = model.StateWriteOnly
case model.StateWriteOnly:
// write only -> delete only
colInfo.State = model.StateDeleteOnly
Expand All @@ -574,15 +573,13 @@ func onDropColumn(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
return ver, errors.Trace(err)
}
job.Args = append(job.Args, indexInfosToIDList(idxInfos))
job.SchemaState = model.StateDeleteOnly
case model.StateDeleteOnly:
// delete only -> reorganization
colInfo.State = model.StateDeleteReorganization
ver, err = updateVersionAndTableInfo(d, t, job, tblInfo, originalState != colInfo.State)
if err != nil {
return ver, errors.Trace(err)
}
job.SchemaState = model.StateDeleteReorganization
case model.StateDeleteReorganization:
// reorganization -> absent
// All reorganization jobs are done, drop this column.
Expand All @@ -604,6 +601,7 @@ func onDropColumn(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
default:
err = dbterror.ErrInvalidDDLJob.GenWithStackByArgs("table", tblInfo.State)
}
job.SchemaState = colInfo.State
return ver, errors.Trace(err)
}

Expand Down
79 changes: 43 additions & 36 deletions ddl/ddl_api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,10 +527,11 @@ func (d *ddl) DropSchema(ctx sessionctx.Context, schema model.CIStr) (err error)
return errors.Trace(infoschema.ErrDatabaseNotExists)
Copy link
Contributor

Choose a reason for hiding this comment

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

Do we need to consider ModifyColumn, TruncateTable, ModifySchemaCharsetAndCollate, DropTablePartition, ExchangeTablePartition, SetDefaultValue, and so on.

Copy link
Contributor Author

@tangenta tangenta Apr 26, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Some of them don't even have the concept of schema state(like truncate table, modify charset/collate), and some involve non-trivial changes(like drop partition, modify columns).

Copy link
Contributor

Choose a reason for hiding this comment

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

How about ActionDropColumns?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

DropColumns will be removed when multi-schema change is supported.

}
job := &model.Job{
SchemaID: old.ID,
SchemaName: old.Name.L,
Type: model.ActionDropSchema,
BinlogInfo: &model.HistoryInfo{},
SchemaID: old.ID,
SchemaName: old.Name.L,
SchemaState: old.State,
Type: model.ActionDropSchema,
BinlogInfo: &model.HistoryInfo{},
}

err = d.DoDDLJob(ctx, job)
Expand Down Expand Up @@ -4032,6 +4033,7 @@ func (d *ddl) DropColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTa
Type: model.ActionDropColumn,
BinlogInfo: &model.HistoryInfo{},
MultiSchemaInfo: multiSchemaInfo,
SchemaState: model.StatePublic,
Args: []interface{}{colName},
}

Expand Down Expand Up @@ -5270,12 +5272,13 @@ func (d *ddl) DropTable(ctx sessionctx.Context, ti ast.Ident) (err error) {
}

job := &model.Job{
SchemaID: schema.ID,
TableID: tb.Meta().ID,
SchemaName: schema.Name.L,
TableName: tb.Meta().Name.L,
Type: model.ActionDropTable,
BinlogInfo: &model.HistoryInfo{},
SchemaID: schema.ID,
TableID: tb.Meta().ID,
SchemaName: schema.Name.L,
SchemaState: schema.State,
TableName: tb.Meta().Name.L,
Type: model.ActionDropTable,
BinlogInfo: &model.HistoryInfo{},
}

err = d.DoDDLJob(ctx, job)
Expand Down Expand Up @@ -5304,12 +5307,13 @@ func (d *ddl) DropView(ctx sessionctx.Context, ti ast.Ident) (err error) {
}

job := &model.Job{
SchemaID: schema.ID,
TableID: tb.Meta().ID,
SchemaName: schema.Name.L,
TableName: tb.Meta().Name.L,
Type: model.ActionDropView,
BinlogInfo: &model.HistoryInfo{},
SchemaID: schema.ID,
TableID: tb.Meta().ID,
SchemaName: schema.Name.L,
SchemaState: tb.Meta().State,
TableName: tb.Meta().Name.L,
Type: model.ActionDropView,
BinlogInfo: &model.HistoryInfo{},
}

err = d.DoDDLJob(ctx, job)
Expand Down Expand Up @@ -5974,13 +5978,14 @@ func (d *ddl) DropForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName model.
}

job := &model.Job{
SchemaID: schema.ID,
TableID: t.Meta().ID,
SchemaName: schema.Name.L,
TableName: t.Meta().Name.L,
Type: model.ActionDropForeignKey,
BinlogInfo: &model.HistoryInfo{},
Args: []interface{}{fkName},
SchemaID: schema.ID,
TableID: t.Meta().ID,
SchemaName: schema.Name.L,
SchemaState: model.StatePublic,
TableName: t.Meta().Name.L,
Type: model.ActionDropForeignKey,
BinlogInfo: &model.HistoryInfo{},
Args: []interface{}{fkName},
}

err = d.DoDDLJob(ctx, job)
Expand Down Expand Up @@ -6030,13 +6035,14 @@ func (d *ddl) DropIndex(ctx sessionctx.Context, ti ast.Ident, indexName model.CI
}

job := &model.Job{
SchemaID: schema.ID,
TableID: t.Meta().ID,
SchemaName: schema.Name.L,
TableName: t.Meta().Name.L,
Type: jobTp,
BinlogInfo: &model.HistoryInfo{},
Args: []interface{}{indexName},
SchemaID: schema.ID,
TableID: t.Meta().ID,
SchemaName: schema.Name.L,
TableName: t.Meta().Name.L,
Type: jobTp,
BinlogInfo: &model.HistoryInfo{},
SchemaState: indexInfo.State,
Args: []interface{}{indexName},
}

err = d.DoDDLJob(ctx, job)
Expand Down Expand Up @@ -6601,12 +6607,13 @@ func (d *ddl) DropSequence(ctx sessionctx.Context, ti ast.Ident, ifExists bool)
}

job := &model.Job{
SchemaID: schema.ID,
TableID: tbl.Meta().ID,
SchemaName: schema.Name.L,
TableName: tbl.Meta().Name.L,
Type: model.ActionDropSequence,
BinlogInfo: &model.HistoryInfo{},
SchemaID: schema.ID,
TableID: tbl.Meta().ID,
SchemaName: schema.Name.L,
SchemaState: tbl.Meta().State,
TableName: tbl.Meta().Name.L,
Type: model.ActionDropSequence,
BinlogInfo: &model.HistoryInfo{},
}

err = d.DoDDLJob(ctx, job)
Expand Down
2 changes: 2 additions & 0 deletions ddl/foreign_key.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,8 +106,10 @@ func onDropForeignKey(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ err
}
// Finish this job.
job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
job.SchemaState = fkInfo.State
return ver, nil
default:
job.SchemaState = fkInfo.State
Copy link
Contributor

Choose a reason for hiding this comment

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

Here we needn't do this?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Updated.

return ver, dbterror.ErrInvalidDDLState.GenWithStackByArgs("foreign key", fkInfo.State)
}

Expand Down
6 changes: 2 additions & 4 deletions ddl/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -647,33 +647,30 @@ func onDropIndex(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
if err != nil {
return ver, errors.Trace(err)
}
job.SchemaState = model.StateWriteOnly
case model.StateWriteOnly:
// write only -> delete only
indexInfo.State = model.StateDeleteOnly
ver, err = updateVersionAndTableInfo(d, t, job, tblInfo, originalState != indexInfo.State)
if err != nil {
return ver, errors.Trace(err)
}
job.SchemaState = model.StateDeleteOnly
case model.StateDeleteOnly:
// delete only -> reorganization
indexInfo.State = model.StateDeleteReorganization
ver, err = updateVersionAndTableInfo(d, t, job, tblInfo, originalState != indexInfo.State)
if err != nil {
return ver, errors.Trace(err)
}
job.SchemaState = model.StateDeleteReorganization
case model.StateDeleteReorganization:
// reorganization -> absent
indexInfo.State = model.StateNone
if len(dependentHiddenCols) > 0 {
firstHiddenOffset := dependentHiddenCols[0].Offset
for i := 0; i < len(dependentHiddenCols); i++ {
// Set this column's offset to the last and reset all following columns' offsets.
adjustColumnInfoInDropColumn(tblInfo, firstHiddenOffset)
}
}

newIndices := make([]*model.IndexInfo, 0, len(tblInfo.Indices))
for _, idx := range tblInfo.Indices {
if idx.Name.L != indexInfo.Name.L {
Expand Down Expand Up @@ -709,6 +706,7 @@ func onDropIndex(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
default:
err = dbterror.ErrInvalidDDLState.GenWithStackByArgs("index", indexInfo.State)
Copy link
Contributor

Choose a reason for hiding this comment

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

If we do this change, do we need to do a return here? Otherwise, there is an error, but also change the schema version?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I think this is fine because the DDL state is incorrect anyway.

job.SchemaState is exactly used to show the indexInfo.State here.

Copy link
Contributor

Choose a reason for hiding this comment

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

However, this may cause a problem if it is later written to KV (DDL job should be retried).

}
job.SchemaState = indexInfo.State
return ver, errors.Trace(err)
}

Expand Down
14 changes: 8 additions & 6 deletions ddl/rollingback.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,7 +387,7 @@ func rollingbackDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, err
if err != nil {
return ver, errors.Trace(err)
}
return cancelOnlyNotHandledJob(job)
return cancelOnlyNotHandledJob(job, model.StateNone)
}

func rollingbackDropSchema(t *meta.Meta, job *model.Job) error {
Expand Down Expand Up @@ -420,9 +420,9 @@ func rollingbackRenameIndex(t *meta.Meta, job *model.Job) (ver int64, err error)
return ver, errors.Trace(err)
}

func cancelOnlyNotHandledJob(job *model.Job) (ver int64, err error) {
func cancelOnlyNotHandledJob(job *model.Job, initialState model.SchemaState) (ver int64, err error) {
// We can only cancel the not handled job.
if job.SchemaState == model.StateNone {
if job.SchemaState == initialState {
job.State = model.JobStateCancelled
return ver, dbterror.ErrCancelledDDLJob
}
Expand All @@ -437,7 +437,7 @@ func rollingbackTruncateTable(t *meta.Meta, job *model.Job) (ver int64, err erro
if err != nil {
return ver, errors.Trace(err)
}
return cancelOnlyNotHandledJob(job)
return cancelOnlyNotHandledJob(job, model.StateNone)
}

func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error) {
Expand Down Expand Up @@ -472,13 +472,15 @@ func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job)
ver, err = rollingbackTruncateTable(t, job)
case model.ActionModifyColumn:
ver, err = rollingbackModifyColumn(w, d, t, job)
case model.ActionDropForeignKey:
ver, err = cancelOnlyNotHandledJob(job, model.StatePublic)
case model.ActionRebaseAutoID, model.ActionShardRowID, model.ActionAddForeignKey,
model.ActionDropForeignKey, model.ActionRenameTable, model.ActionRenameTables,
model.ActionRenameTable, model.ActionRenameTables,
model.ActionModifyTableCharsetAndCollate, model.ActionTruncateTablePartition,
model.ActionModifySchemaCharsetAndCollate, model.ActionRepairTable,
model.ActionModifyTableAutoIdCache, model.ActionAlterIndexVisibility,
model.ActionExchangeTablePartition, model.ActionModifySchemaDefaultPlacement:
ver, err = cancelOnlyNotHandledJob(job)
ver, err = cancelOnlyNotHandledJob(job, model.StateNone)
default:
job.State = model.JobStateCancelled
err = dbterror.ErrCancelledDDLJob
Expand Down
Loading