diff --git a/ddl/db_test.go b/ddl/db_test.go index a5e0810578414..1406e0b639370 100644 --- a/ddl/db_test.go +++ b/ddl/db_test.go @@ -47,6 +47,7 @@ import ( "github.com/pingcap/tidb/types" "github.com/pingcap/tidb/util/admin" "github.com/pingcap/tidb/util/mock" + "github.com/pingcap/tidb/util/schemautil" "github.com/pingcap/tidb/util/testkit" "github.com/pingcap/tidb/util/testleak" "github.com/pingcap/tidb/util/testutil" @@ -518,6 +519,96 @@ func (s *testDBSuite) TestCancelAddIndex1(c *C) { s.mustExec(c, "alter table t drop index idx_c2") } +// TestCancelDropIndex tests cancel ddl job which type is drop index. +func (s *testDBSuite) TestCancelDropIndex(c *C) { + s.tk = testkit.NewTestKit(c, s.store) + s.mustExec(c, "use test_db") + s.mustExec(c, "drop table if exists t") + s.mustExec(c, "create table t(c1 int, c2 int)") + defer s.mustExec(c, "drop table t;") + for i := 0; i < 5; i++ { + s.mustExec(c, "insert into t values (?, ?)", i, i) + } + + testCases := []struct { + needAddIndex bool + jobState model.JobState + JobSchemaState model.SchemaState + cancelSucc bool + }{ + // model.JobStateNone means the jobs is canceled before the first run. + {true, model.JobStateNone, model.StateNone, true}, + {false, model.JobStateRunning, model.StateWriteOnly, true}, + {false, model.JobStateRunning, model.StateDeleteOnly, false}, + {true, model.JobStateRunning, model.StateDeleteReorganization, false}, + } + + var checkErr error + oldReorgWaitTimeout := ddl.ReorgWaitTimeout + ddl.ReorgWaitTimeout = 50 * time.Millisecond + defer func() { ddl.ReorgWaitTimeout = oldReorgWaitTimeout }() + hook := &ddl.TestDDLCallback{} + var jobID int64 + testCase := &testCases[0] + hook.OnJobRunBeforeExported = func(job *model.Job) { + if job.Type == model.ActionDropIndex && job.State == testCase.jobState && job.SchemaState == testCase.JobSchemaState { + jobID = job.ID + jobIDs := []int64{job.ID} + hookCtx := mock.NewContext() + hookCtx.Store = s.store + err := hookCtx.NewTxn(context.TODO()) + if err != nil { + checkErr = errors.Trace(err) + return + } + errs, err := admin.CancelJobs(hookCtx.Txn(true), jobIDs) + if err != nil { + checkErr = errors.Trace(err) + return + } + + if errs[0] != nil { + checkErr = errors.Trace(errs[0]) + return + } + + checkErr = hookCtx.Txn(true).Commit(context.Background()) + } + } + s.dom.DDL().(ddl.DDLForTest).SetHook(hook) + for i := range testCases { + testCase = &testCases[i] + if testCase.needAddIndex { + s.mustExec(c, "alter table t add index idx_c2(c2)") + } + rs, err := s.tk.Exec("alter table t drop index idx_c2") + if rs != nil { + rs.Close() + } + + t := s.testGetTable(c, "t") + indexInfo := schemautil.FindIndexByName("idx_c2", t.Meta().Indices) + if testCase.cancelSucc { + c.Assert(checkErr, IsNil) + c.Assert(err, NotNil) + c.Assert(err.Error(), Equals, "[ddl:12]cancelled DDL job") + + c.Assert(indexInfo, NotNil) + c.Assert(indexInfo.State, Equals, model.StatePublic) + } else { + err1 := admin.ErrCannotCancelDDLJob.GenWithStackByArgs(jobID) + c.Assert(err, IsNil) + c.Assert(checkErr, NotNil) + c.Assert(checkErr.Error(), Equals, err1.Error()) + + c.Assert(indexInfo, IsNil) + } + } + s.dom.DDL().(ddl.DDLForTest).SetHook(&ddl.TestDDLCallback{}) + s.mustExec(c, "alter table t add index idx_c2(c2)") + s.mustExec(c, "alter table t drop index idx_c2") +} + func (s *testDBSuite) TestAddAnonymousIndex(c *C) { s.tk = testkit.NewTestKit(c, s.store) s.tk.MustExec("use " + s.schemaName) diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go index c51972f6b7506..ad1c05deeec7d 100644 --- a/ddl/ddl_api.go +++ b/ddl/ddl_api.go @@ -36,6 +36,7 @@ import ( "github.com/pingcap/tidb/sessionctx/variable" "github.com/pingcap/tidb/table" "github.com/pingcap/tidb/types" + "github.com/pingcap/tidb/util/schemautil" ) func (d *ddl) CreateSchema(ctx sessionctx.Context, schema model.CIStr, charsetInfo *ast.CharsetOpt) (err error) { @@ -2027,7 +2028,7 @@ func (d *ddl) CreateIndex(ctx sessionctx.Context, ti ast.Ident, unique bool, ind indexName = getAnonymousIndex(t, idxColNames[0].Column.Name) } - if indexInfo := findIndexByName(indexName.L, t.Meta().Indices); indexInfo != nil { + if indexInfo := schemautil.FindIndexByName(indexName.L, t.Meta().Indices); indexInfo != nil { return ErrDupKeyName.GenWithStack("index already exist %s", indexName) } @@ -2162,7 +2163,7 @@ func (d *ddl) DropIndex(ctx sessionctx.Context, ti ast.Ident, indexName model.CI return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ti.Schema, ti.Name)) } - if indexInfo := findIndexByName(indexName.L, t.Meta().Indices); indexInfo == nil { + if indexInfo := schemautil.FindIndexByName(indexName.L, t.Meta().Indices); indexInfo == nil { return ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName) } diff --git a/ddl/ddl_worker_test.go b/ddl/ddl_worker_test.go index 435e428c0a7af..3eaefcaed68e7 100644 --- a/ddl/ddl_worker_test.go +++ b/ddl/ddl_worker_test.go @@ -331,7 +331,6 @@ type testCancelJob struct { func buildCancelJobTests(firstID int64) []testCancelJob { err := errCancelledDDLJob - errs := []error{err} noErrs := []error{nil} tests := []testCancelJob{ {act: model.ActionAddIndex, jobIDs: []int64{firstID + 1}, cancelRetErrs: noErrs, cancelState: model.StateDeleteOnly, ddlRetErr: err}, @@ -339,18 +338,15 @@ func buildCancelJobTests(firstID int64) []testCancelJob { {act: model.ActionAddIndex, jobIDs: []int64{firstID + 3}, cancelRetErrs: noErrs, cancelState: model.StateWriteReorganization, ddlRetErr: err}, {act: model.ActionAddIndex, jobIDs: []int64{firstID + 4}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 4)}, cancelState: model.StatePublic, ddlRetErr: err}, - // TODO: after fix drop index and create table rollback bug, the below test cases maybe need to change. - {act: model.ActionDropIndex, jobIDs: []int64{firstID + 5}, cancelRetErrs: errs, cancelState: model.StateWriteOnly, ddlRetErr: err}, - {act: model.ActionDropIndex, jobIDs: []int64{firstID + 6}, cancelRetErrs: errs, cancelState: model.StateDeleteOnly, ddlRetErr: err}, - {act: model.ActionDropIndex, jobIDs: []int64{firstID + 7}, cancelRetErrs: errs, cancelState: model.StateDeleteReorganization, ddlRetErr: err}, - {act: model.ActionDropIndex, jobIDs: []int64{firstID + 8}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 8)}, cancelState: model.StateNone, ddlRetErr: err}, + // Test cancel drop index job , see TestCancelDropIndex. + // TODO: add create table back after we fix the cancel bug. //{act: model.ActionCreateTable, jobIDs: []int64{firstID + 9}, cancelRetErrs: noErrs, cancelState: model.StatePublic, ddlRetErr: err}, - {act: model.ActionAddColumn, jobIDs: []int64{firstID + 9}, cancelRetErrs: noErrs, cancelState: model.StateDeleteOnly, ddlRetErr: err}, - {act: model.ActionAddColumn, jobIDs: []int64{firstID + 10}, cancelRetErrs: noErrs, cancelState: model.StateWriteOnly, ddlRetErr: err}, - {act: model.ActionAddColumn, jobIDs: []int64{firstID + 11}, cancelRetErrs: noErrs, cancelState: model.StateWriteReorganization, ddlRetErr: err}, - {act: model.ActionAddColumn, jobIDs: []int64{firstID + 12}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 12)}, cancelState: model.StatePublic, ddlRetErr: err}, + {act: model.ActionAddColumn, jobIDs: []int64{firstID + 5}, cancelRetErrs: noErrs, cancelState: model.StateDeleteOnly, ddlRetErr: err}, + {act: model.ActionAddColumn, jobIDs: []int64{firstID + 6}, cancelRetErrs: noErrs, cancelState: model.StateWriteOnly, ddlRetErr: err}, + {act: model.ActionAddColumn, jobIDs: []int64{firstID + 7}, cancelRetErrs: noErrs, cancelState: model.StateWriteReorganization, ddlRetErr: err}, + {act: model.ActionAddColumn, jobIDs: []int64{firstID + 8}, cancelRetErrs: []error{admin.ErrCancelFinishedDDLJob.GenWithStackByArgs(firstID + 8)}, cancelState: model.StatePublic, ddlRetErr: err}, } return tests @@ -464,23 +460,8 @@ func (s *testDDLSuite) TestCancelJob(c *C) { c.Assert(ctx.Txn(true).Commit(context.Background()), IsNil) s.checkAddIdx(c, d, dbInfo.ID, tblInfo.ID, idxOrigName, true) - // for dropping index - idxName := []interface{}{model.NewCIStr("idx")} - test = &tests[4] - doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, tblInfo.ID, model.ActionDropIndex, idxName, &test.cancelState) - c.Check(errors.ErrorStack(checkErr), Equals, "") - test = &tests[5] - doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, tblInfo.ID, model.ActionDropIndex, idxName, &test.cancelState) - c.Check(errors.ErrorStack(checkErr), Equals, "") - test = &tests[6] - doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, tblInfo.ID, model.ActionDropIndex, idxName, &test.cancelState) - c.Check(errors.ErrorStack(checkErr), Equals, "") - test = &tests[7] - testDropIndex(c, ctx, d, dbInfo, tblInfo, "idx") - c.Check(errors.ErrorStack(checkErr), Equals, "") - // for add column - test = &tests[8] + test = &tests[4] addingColName := "colA" newColumnDef := &ast.ColumnDef{ @@ -494,17 +475,17 @@ func (s *testDDLSuite) TestCancelJob(c *C) { c.Check(errors.ErrorStack(checkErr), Equals, "") s.checkAddColumn(c, d, dbInfo.ID, tblInfo.ID, addingColName, false) - test = &tests[9] + test = &tests[5] doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, tblInfo.ID, model.ActionAddColumn, addColumnArgs, &cancelState) c.Check(errors.ErrorStack(checkErr), Equals, "") s.checkAddColumn(c, d, dbInfo.ID, tblInfo.ID, addingColName, false) - test = &tests[10] + test = &tests[6] doDDLJobErrWithSchemaState(ctx, d, c, dbInfo.ID, tblInfo.ID, model.ActionAddColumn, addColumnArgs, &cancelState) c.Check(errors.ErrorStack(checkErr), Equals, "") s.checkAddColumn(c, d, dbInfo.ID, tblInfo.ID, addingColName, false) - test = &tests[11] + test = &tests[7] testAddColumn(c, ctx, d, dbInfo, tblInfo, addColumnArgs) c.Check(errors.ErrorStack(checkErr), Equals, "") s.checkAddColumn(c, d, dbInfo.ID, tblInfo.ID, addingColName, true) diff --git a/ddl/index.go b/ddl/index.go index a97a1c32cb337..e1aefee387c0f 100644 --- a/ddl/index.go +++ b/ddl/index.go @@ -37,6 +37,7 @@ import ( "github.com/pingcap/tidb/types" "github.com/pingcap/tidb/util" "github.com/pingcap/tidb/util/rowDecoder" + "github.com/pingcap/tidb/util/schemautil" "github.com/pingcap/tidb/util/timeutil" log "github.com/sirupsen/logrus" ) @@ -182,7 +183,7 @@ func dropIndexColumnFlag(tblInfo *model.TableInfo, indexInfo *model.IndexInfo) { } func validateRenameIndex(from, to model.CIStr, tbl *model.TableInfo) (ignore bool, err error) { - if fromIdx := findIndexByName(from.L, tbl.Indices); fromIdx == nil { + if fromIdx := schemautil.FindIndexByName(from.L, tbl.Indices); fromIdx == nil { return false, errors.Trace(infoschema.ErrKeyNotExists.GenWithStackByArgs(from.O, tbl.Name)) } // Take case-sensitivity into account, if `FromKey` and `ToKey` are the same, nothing need to be changed @@ -192,7 +193,7 @@ func validateRenameIndex(from, to model.CIStr, tbl *model.TableInfo) (ignore boo // If spec.FromKey.L == spec.ToKey.L, we operate on the same index(case-insensitive) and change its name (case-sensitive) // e.g: from `inDex` to `IndEX`. Otherwise, we try to rename an index to another different index which already exists, // that's illegal by rule. - if toIdx := findIndexByName(to.L, tbl.Indices); toIdx != nil && from.L != to.L { + if toIdx := schemautil.FindIndexByName(to.L, tbl.Indices); toIdx != nil && from.L != to.L { return false, errors.Trace(infoschema.ErrKeyNameDuplicate.GenWithStackByArgs(toIdx.Name.O)) } return false, nil @@ -219,7 +220,7 @@ func onRenameIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) { job.State = model.JobStateCancelled return ver, errors.Trace(err) } - idx := findIndexByName(from.L, tblInfo.Indices) + idx := schemautil.FindIndexByName(from.L, tblInfo.Indices) idx.Name = to if ver, err = updateVersionAndTableInfo(t, job, tblInfo, true); err != nil { job.State = model.JobStateCancelled @@ -258,7 +259,7 @@ func (w *worker) onCreateIndex(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int return ver, errors.Trace(err) } - indexInfo := findIndexByName(indexName.L, tblInfo.Indices) + indexInfo := schemautil.FindIndexByName(indexName.L, tblInfo.Indices) if indexInfo != nil && indexInfo.State == model.StatePublic { job.State = model.JobStateCancelled return ver, ErrDupKeyName.GenWithStack("index already exist %s", indexName) @@ -371,7 +372,7 @@ func onDropIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) { return ver, errors.Trace(err) } - indexInfo := findIndexByName(indexName.L, tblInfo.Indices) + indexInfo := schemautil.FindIndexByName(indexName.L, tblInfo.Indices) if indexInfo == nil { job.State = model.JobStateCancelled return ver, ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName) @@ -1174,15 +1175,6 @@ func findNextPartitionID(currentPartition int64, defs []model.PartitionDefinitio return 0, errors.Errorf("partition id not found %d", currentPartition) } -func findIndexByName(idxName string, indices []*model.IndexInfo) *model.IndexInfo { - for _, idx := range indices { - if idx.Name.L == idxName { - return idx - } - } - return nil -} - func allocateIndexID(tblInfo *model.TableInfo) int64 { tblInfo.MaxIndexID++ return tblInfo.MaxIndexID diff --git a/ddl/rollingback.go b/ddl/rollingback.go index 74f9513a6b0c9..db8da52003162 100644 --- a/ddl/rollingback.go +++ b/ddl/rollingback.go @@ -20,6 +20,7 @@ import ( "github.com/pingcap/tidb/infoschema" "github.com/pingcap/tidb/kv" "github.com/pingcap/tidb/meta" + "github.com/pingcap/tidb/util/schemautil" log "github.com/sirupsen/logrus" ) @@ -67,7 +68,7 @@ func convertNotStartAddIdxJob2RollbackJob(t *meta.Meta, job *model.Job, occuredE return ver, errors.Trace(err) } - indexInfo := findIndexByName(indexName.L, tblInfo.Indices) + indexInfo := schemautil.FindIndexByName(indexName.L, tblInfo.Indices) if indexInfo == nil { job.State = model.JobStateCancelled return ver, errCancelledDDLJob @@ -117,6 +118,50 @@ func rollingbackAddColumn(t *meta.Meta, job *model.Job) (ver int64, err error) { return ver, errCancelledDDLJob } +func rollingbackDropIndex(t *meta.Meta, job *model.Job) (ver int64, err error) { + schemaID := job.SchemaID + tblInfo, err := getTableInfo(t, job, schemaID) + if err != nil { + return ver, errors.Trace(err) + } + + var indexName model.CIStr + err = job.DecodeArgs(&indexName) + if err != nil { + job.State = model.JobStateCancelled + return ver, errors.Trace(err) + } + + indexInfo := schemautil.FindIndexByName(indexName.L, tblInfo.Indices) + if indexInfo == nil { + job.State = model.JobStateCancelled + return ver, ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName) + } + + originalState := indexInfo.State + switch indexInfo.State { + case model.StateDeleteOnly, model.StateDeleteReorganization, model.StateNone: + // We can not rollback now, so just continue to drop index. + // Normally won't fetch here, because there is check when cancel ddl jobs. see function: isJobRollbackable. + job.State = model.JobStateRunning + return ver, nil + case model.StatePublic, model.StateWriteOnly: + job.State = model.JobStateRollbackDone + indexInfo.State = model.StatePublic + default: + return ver, ErrInvalidIndexState.GenWithStack("invalid index state %v", indexInfo.State) + } + + job.SchemaState = indexInfo.State + job.Args = []interface{}{indexInfo.Name} + ver, err = updateVersionAndTableInfo(t, job, tblInfo, originalState != indexInfo.State) + if err != nil { + return ver, errors.Trace(err) + } + job.FinishTableJob(model.JobStateRollbackDone, model.StatePublic, ver, tblInfo) + return ver, errCancelledDDLJob +} + func rollingbackAddindex(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error) { // If the value of SnapshotVer isn't zero, it means the work is backfilling the indexes. if job.SchemaState == model.StateWriteReorganization && job.SnapshotVer != 0 { @@ -137,6 +182,8 @@ func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) ver, err = rollingbackAddColumn(t, job) case model.ActionAddIndex: ver, err = rollingbackAddindex(w, d, t, job) + case model.ActionDropIndex: + ver, err = rollingbackDropIndex(t, job) default: job.State = model.JobStateCancelled err = errCancelledDDLJob diff --git a/planner/core/planbuilder.go b/planner/core/planbuilder.go index cbf08c31c30a4..6faad83705589 100644 --- a/planner/core/planbuilder.go +++ b/planner/core/planbuilder.go @@ -33,6 +33,7 @@ import ( "github.com/pingcap/tidb/types" "github.com/pingcap/tidb/types/parser_driver" "github.com/pingcap/tidb/util/ranger" + "github.com/pingcap/tidb/util/schemautil" ) type visitInfo struct { @@ -383,15 +384,6 @@ func removeIgnoredPaths(paths, ignoredPaths []*accessPath, tblInfo *model.TableI return remainedPaths } -func findIndexByName(indices []*model.IndexInfo, name model.CIStr) *model.IndexInfo { - for _, idx := range indices { - if idx.Name.L == name.L { - return idx - } - } - return nil -} - func (b *PlanBuilder) buildSelectLock(src LogicalPlan, lock ast.SelectLockType) *LogicalLock { selectLock := LogicalLock{Lock: lock}.Init(b.ctx) selectLock.SetChildren(src) @@ -723,7 +715,7 @@ func (b *PlanBuilder) buildAnalyzeIndex(as *ast.AnalyzeTableStmt) (Plan, error) return nil, err } for _, idxName := range as.IndexNames { - idx := findIndexByName(tblInfo.Indices, idxName) + idx := schemautil.FindIndexByName(idxName.L, tblInfo.Indices) if idx == nil || idx.State != model.StatePublic { return nil, ErrAnalyzeMissIndex.GenWithStackByArgs(idxName.O, tblInfo.Name.O) } diff --git a/util/admin/admin.go b/util/admin/admin.go index 1c4fa7107a0fd..27838ca2cadc2 100644 --- a/util/admin/admin.go +++ b/util/admin/admin.go @@ -83,6 +83,18 @@ func GetDDLInfo(txn kv.Transaction) (*DDLInfo, error) { return info, nil } +func isJobRollbackable(job *model.Job, id int64) error { + switch job.Type { + case model.ActionDropIndex: + // We can't cancel if index current state is in StateDeleteOnly or StateDeleteReorganization, otherwise will cause inconsistent between record and index. + if job.SchemaState == model.StateDeleteOnly || + job.SchemaState == model.StateDeleteReorganization { + return ErrCannotCancelDDLJob.GenWithStackByArgs(id) + } + } + return nil +} + // CancelJobs cancels the DDL jobs. func CancelJobs(txn kv.Transaction, ids []int64) ([]error, error) { if len(ids) == 0 { @@ -113,6 +125,11 @@ func CancelJobs(txn kv.Transaction, ids []int64) ([]error, error) { if job.IsCancelled() || job.IsRollingback() || job.IsRollbackDone() { continue } + errs[i] = isJobRollbackable(job, id) + if errs[i] != nil { + continue + } + job.State = model.JobStateCancelling // Make sure RawArgs isn't overwritten. err := job.DecodeArgs(job.RawArgs) @@ -701,6 +718,7 @@ const ( codeInvalidColumnState = 3 codeDDLJobNotFound = 4 codeCancelFinishedJob = 5 + codeCannotCancelDDLJob = 6 ) var ( @@ -712,4 +730,6 @@ var ( ErrDDLJobNotFound = terror.ClassAdmin.New(codeDDLJobNotFound, "DDL Job:%v not found") // ErrCancelFinishedDDLJob returns when cancel a finished ddl job. ErrCancelFinishedDDLJob = terror.ClassAdmin.New(codeCancelFinishedJob, "This job:%v is finished, so can't be cancelled") + // ErrCannotCancelDDLJob returns when cancel a almost finished ddl job, because cancel in now may cause data inconsistency. + ErrCannotCancelDDLJob = terror.ClassAdmin.New(codeCannotCancelDDLJob, "This job:%v is almost finished, can't be cancelled now") ) diff --git a/util/schemautil/schemautil.go b/util/schemautil/schemautil.go new file mode 100644 index 0000000000000..6a12b5ead734d --- /dev/null +++ b/util/schemautil/schemautil.go @@ -0,0 +1,26 @@ +// Copyright 2018 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package schemautil + +import "github.com/pingcap/parser/model" + +// FindIndexByName finds index by name. +func FindIndexByName(idxName string, indices []*model.IndexInfo) *model.IndexInfo { + for _, idx := range indices { + if idx.Name.L == idxName { + return idx + } + } + return nil +}