// 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 ddl import ( "github.com/pingcap/errors" "github.com/pingcap/parser/ast" "github.com/pingcap/parser/model" "github.com/pingcap/parser/mysql" "github.com/pingcap/tidb/kv" "github.com/pingcap/tidb/meta" "github.com/pingcap/tidb/util/logutil" "go.uber.org/zap" ) func updateColsNull2NotNull(tblInfo *model.TableInfo, indexInfo *model.IndexInfo) error { nullCols, err := getNullColInfos(tblInfo, indexInfo) if err != nil { return errors.Trace(err) } for _, col := range nullCols { col.Flag |= mysql.NotNullFlag col.Flag = col.Flag &^ mysql.PreventNullInsertFlag } return nil } func convertAddIdxJob2RollbackJob(t *meta.Meta, job *model.Job, tblInfo *model.TableInfo, indexInfo *model.IndexInfo, err error) (int64, error) { job.State = model.JobStateRollingback if indexInfo.Primary { nullCols, err := getNullColInfos(tblInfo, indexInfo) if err != nil { return 0, errors.Trace(err) } for _, col := range nullCols { // Field PreventNullInsertFlag flag reset. col.Flag = col.Flag &^ mysql.PreventNullInsertFlag } } // the second args will be used in onDropIndex. job.Args = []interface{}{indexInfo.Name, getPartitionIDs(tblInfo)} // If add index job rollbacks in write reorganization state, its need to delete all keys which has been added. // Its work is the same as drop index job do. // The write reorganization state in add index job that likes write only state in drop index job. // So the next state is delete only state. originalState := indexInfo.State indexInfo.State = model.StateDeleteOnly // Change dependent hidden columns if necessary. updateHiddenColumns(tblInfo, indexInfo, model.StateDeleteOnly) job.SchemaState = model.StateDeleteOnly ver, err1 := updateVersionAndTableInfo(t, job, tblInfo, originalState != indexInfo.State) if err1 != nil { return ver, errors.Trace(err1) } if kv.ErrKeyExists.Equal(err) { return ver, kv.ErrKeyExists.GenWithStackByArgs("", indexInfo.Name.O) } return ver, errors.Trace(err) } // convertNotStartAddIdxJob2RollbackJob converts the add index job that are not started workers to rollingbackJob, // to rollback add index operations. job.SnapshotVer == 0 indicates the workers are not started. func convertNotStartAddIdxJob2RollbackJob(t *meta.Meta, job *model.Job, occuredErr error) (ver int64, err error) { schemaID := job.SchemaID tblInfo, err := getTableInfoAndCancelFaultJob(t, job, schemaID) if err != nil { return ver, errors.Trace(err) } var ( unique bool indexName model.CIStr indexPartSpecifications []*ast.IndexPartSpecification indexOption *ast.IndexOption ) err = job.DecodeArgs(&unique, &indexName, &indexPartSpecifications, &indexOption) if err != nil { job.State = model.JobStateCancelled return ver, errors.Trace(err) } indexInfo := tblInfo.FindIndexByName(indexName.L) if indexInfo == nil { job.State = model.JobStateCancelled return ver, errCancelledDDLJob } return convertAddIdxJob2RollbackJob(t, job, tblInfo, indexInfo, occuredErr) } func rollingbackAddColumn(t *meta.Meta, job *model.Job) (ver int64, err error) { job.State = model.JobStateRollingback tblInfo, columnInfo, col, _, _, err := checkAddColumn(t, job) if err != nil { return ver, errors.Trace(err) } if columnInfo == nil { job.State = model.JobStateCancelled return ver, errCancelledDDLJob } originalState := columnInfo.State columnInfo.State = model.StateDeleteOnly job.SchemaState = model.StateDeleteOnly job.Args = []interface{}{col.Name} ver, err = updateVersionAndTableInfo(t, job, tblInfo, originalState != columnInfo.State) if err != nil { return ver, errors.Trace(err) } return ver, errCancelledDDLJob } func rollingbackDropColumn(t *meta.Meta, job *model.Job) (ver int64, err error) { tblInfo, colInfo, err := checkDropColumn(t, job) if err != nil { return ver, errors.Trace(err) } // StatePublic means when the job is not running yet. if colInfo.State == model.StatePublic { job.State = model.JobStateCancelled job.FinishTableJob(model.JobStateRollbackDone, model.StatePublic, ver, tblInfo) return ver, errCancelledDDLJob } // In the state of drop column `write only -> delete only -> reorganization`, // We can not rollback now, so just continue to drop column. job.State = model.JobStateRunning return ver, nil } func rollingbackDropIndex(t *meta.Meta, job *model.Job) (ver int64, err error) { tblInfo, indexInfo, err := checkDropIndex(t, job) if err != nil { return ver, errors.Trace(err) } originalState := indexInfo.State switch indexInfo.State { case model.StateWriteOnly, 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: job.State = model.JobStateRollbackDone indexInfo.State = model.StatePublic default: return ver, ErrInvalidDDLState.GenWithStackByArgs("index", 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, isPK bool) (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 { // add index workers are started. need to ask them to exit. logutil.Logger(w.logCtx).Info("[ddl] run the cancelling DDL job", zap.String("job", job.String())) w.reorgCtx.notifyReorgCancel() ver, err = w.onCreateIndex(d, t, job, isPK) } else { // add index workers are not started, remove the indexInfo in tableInfo. ver, err = convertNotStartAddIdxJob2RollbackJob(t, job, errCancelledDDLJob) } return } func convertAddTablePartitionJob2RollbackJob(t *meta.Meta, job *model.Job, otherwiseErr error, tblInfo *model.TableInfo) (ver int64, err error) { job.State = model.JobStateRollingback addingDefinitions := tblInfo.Partition.AddingDefinitions partNames := make([]string, 0, len(addingDefinitions)) for _, pd := range addingDefinitions { partNames = append(partNames, pd.Name.L) } job.Args = []interface{}{partNames} ver, err = updateVersionAndTableInfo(t, job, tblInfo, true) if err != nil { return ver, errors.Trace(err) } return ver, errors.Trace(otherwiseErr) } func rollingbackAddTablePartition(t *meta.Meta, job *model.Job) (ver int64, err error) { tblInfo, _, addingDefinitions, err := checkAddPartition(t, job) if err != nil { return ver, errors.Trace(err) } // addingDefinitions' len = 0 means the job hasn't reached the replica-only state. if len(addingDefinitions) == 0 { job.State = model.JobStateCancelled return ver, errors.Trace(errCancelledDDLJob) } // addingDefinitions is also in tblInfo, here pass the tblInfo as parameter directly. return convertAddTablePartitionJob2RollbackJob(t, job, errCancelledDDLJob, tblInfo) } func rollingbackDropTableOrView(t *meta.Meta, job *model.Job) error { tblInfo, err := checkTableExistAndCancelNonExistJob(t, job, job.SchemaID) if err != nil { return errors.Trace(err) } // To simplify the rollback logic, cannot be canceled after job start to run. // Normally won't fetch here, because there is check when cancel ddl jobs. see function: isJobRollbackable. if tblInfo.State == model.StatePublic { job.State = model.JobStateCancelled return errCancelledDDLJob } job.State = model.JobStateRunning return nil } func rollingbackDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, err error) { _, err = getTableInfoAndCancelFaultJob(t, job, job.SchemaID) if err != nil { return ver, errors.Trace(err) } return cancelOnlyNotHandledJob(job) } func rollingbackDropSchema(t *meta.Meta, job *model.Job) error { dbInfo, err := checkSchemaExistAndCancelNotExistJob(t, job) if err != nil { return errors.Trace(err) } // To simplify the rollback logic, cannot be canceled after job start to run. // Normally won't fetch here, because there is check when cancel ddl jobs. see function: isJobRollbackable. if dbInfo.State == model.StatePublic { job.State = model.JobStateCancelled return errCancelledDDLJob } job.State = model.JobStateRunning return nil } func rollingbackRenameIndex(t *meta.Meta, job *model.Job) (ver int64, err error) { tblInfo, from, _, err := checkRenameIndex(t, job) if err != nil { return ver, errors.Trace(err) } // Here rename index is done in a transaction, if the job is not completed, it can be canceled. idx := tblInfo.FindIndexByName(from.L) if idx.State == model.StatePublic { job.State = model.JobStateCancelled return ver, errCancelledDDLJob } job.State = model.JobStateRunning return ver, errors.Trace(err) } func cancelOnlyNotHandledJob(job *model.Job) (ver int64, err error) { // We can only cancel the not handled job. if job.SchemaState == model.StateNone { job.State = model.JobStateCancelled return ver, errCancelledDDLJob } job.State = model.JobStateRunning return ver, nil } func rollingbackTruncateTable(t *meta.Meta, job *model.Job) (ver int64, err error) { _, err = getTableInfoAndCancelFaultJob(t, job, job.SchemaID) if err != nil { return ver, errors.Trace(err) } return cancelOnlyNotHandledJob(job) } func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error) { switch job.Type { case model.ActionAddColumn: ver, err = rollingbackAddColumn(t, job) case model.ActionAddIndex: ver, err = rollingbackAddIndex(w, d, t, job, false) case model.ActionAddPrimaryKey: ver, err = rollingbackAddIndex(w, d, t, job, true) case model.ActionAddTablePartition: ver, err = rollingbackAddTablePartition(t, job) case model.ActionDropColumn: ver, err = rollingbackDropColumn(t, job) case model.ActionDropIndex, model.ActionDropPrimaryKey: ver, err = rollingbackDropIndex(t, job) case model.ActionDropTable, model.ActionDropView, model.ActionDropSequence: err = rollingbackDropTableOrView(t, job) case model.ActionDropTablePartition: ver, err = rollingbackDropTablePartition(t, job) case model.ActionDropSchema: err = rollingbackDropSchema(t, job) case model.ActionRenameIndex: ver, err = rollingbackRenameIndex(t, job) case model.ActionTruncateTable: ver, err = rollingbackTruncateTable(t, job) case model.ActionRebaseAutoID, model.ActionShardRowID, model.ActionModifyColumn, model.ActionAddForeignKey, model.ActionDropForeignKey, model.ActionRenameTable, model.ActionModifyTableCharsetAndCollate, model.ActionTruncateTablePartition, model.ActionModifySchemaCharsetAndCollate, model.ActionRepairTable, model.ActionModifyTableAutoIdCache: ver, err = cancelOnlyNotHandledJob(job) default: job.State = model.JobStateCancelled err = errCancelledDDLJob } if err != nil { if job.State != model.JobStateRollingback && job.State != model.JobStateCancelled { logutil.Logger(w.logCtx).Error("[ddl] run DDL job failed", zap.String("job", job.String()), zap.Error(err)) } else { logutil.Logger(w.logCtx).Info("[ddl] the DDL job is cancelled normally", zap.String("job", job.String()), zap.Error(err)) } job.Error = toTError(err) job.ErrorCount++ } return }