fix integration tests

ddl/column.go

@@ -217,6 +217,9 @@ func onAddColumn(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error)
 }
 
 func locateOffsetForColumn(pos *ast.ColumnPosition, tblInfo *model.TableInfo) (offset int, err error) {
+	if pos == nil {
+		return -1, nil
+	}
 	// Get column offset.
 	switch pos.Tp {
 	case ast.ColumnPositionFirst:
@@ -362,7 +365,7 @@ func onAddColumns(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error
 	case model.StateWriteReorganization:
 		// reorganization -> public
 		// Adjust table column offsets.
-		for i, newCol := range tblInfo.Columns[:len(tblInfo.Columns)-len(positions)] {
+		for i, newCol := range tblInfo.Columns[len(tblInfo.Columns)-len(positions):] {
 			offset, err := locateOffsetForColumn(positions[i], tblInfo)
 			if err != nil {
 				return ver, errors.Trace(err)

ddl/column_modify_test.go

@@ -568,7 +568,10 @@ func TestCancelDropColumns(t *testing.T) {
 	var jobID int64
 	testCase := &testCases[0]
 	hook.OnJobRunBeforeExported = func(job *model.Job) {
-		if job.Type == model.ActionDropColumns && job.State == testCase.jobState && job.SchemaState == testCase.JobSchemaState {
+		isDropColTp := job.Type == model.ActionMultiSchemaChange &&
+			job.MultiSchemaInfo.SubJobs[0].Type == model.ActionDropColumn
+		if isDropColTp && job.MultiSchemaInfo.SubJobs[0].State == testCase.jobState &&
+			job.MultiSchemaInfo.SubJobs[0].SchemaState == testCase.JobSchemaState {
 			jobIDs := []int64{job.ID}
 			jobID = job.ID
 			hookCtx := mock.NewContext()

ddl/ddl.go

@@ -709,14 +709,16 @@ func (d *ddl) doDDLJob(ctx sessionctx.Context, job *model.Job) error {
 			return errors.Trace(historyJob.Error)
 		}
 		// Only for JobStateCancelled job which is adding columns or drop columns or drop indexes.
-		if historyJob.IsCancelled() && (historyJob.Type == model.ActionAddColumns || historyJob.Type == model.ActionDropColumns || historyJob.Type == model.ActionDropIndexes) {
+		if historyJob.IsCancelled() && (historyJob.Type == model.ActionAddColumns || historyJob.Type == model.ActionDropColumns ||
+			historyJob.Type == model.ActionDropIndexes ||
+			historyJob.Type == model.ActionMultiSchemaChange) {
 			if historyJob.MultiSchemaInfo != nil && len(historyJob.MultiSchemaInfo.Warnings) != 0 {
 				for _, warning := range historyJob.MultiSchemaInfo.Warnings {
 					ctx.GetSessionVars().StmtCtx.AppendWarning(warning)
 				}
 			}
 			logutil.BgLogger().Info("[ddl] DDL job is cancelled", zap.Int64("jobID", jobID))
-			return nil
+			return errCancelledDDLJob
 		}
 		panic("When the state is JobStateRollbackDone or JobStateCancelled, historyJob.Error should never be nil")
 	}

ddl/ddl_api.go

@@ -3949,13 +3949,17 @@ func (d *ddl) DropColumn(ctx sessionctx.Context, ti ast.Ident, spec *ast.AlterTa
 		return err
 	}
 
+	var multiSchemaInfo *model.MultiSchemaInfo
+	if ctx.GetSessionVars().EnableChangeMultiSchema {
+		multiSchemaInfo = &model.MultiSchemaInfo{}
+	}
 	job := &model.Job{
 		SchemaID:        schema.ID,
 		TableID:         t.Meta().ID,
 		SchemaName:      schema.Name.L,
 		Type:            model.ActionDropColumn,
 		BinlogInfo:      &model.HistoryInfo{},
-		MultiSchemaInfo: nil,
+		MultiSchemaInfo: multiSchemaInfo,
 		Args:            []interface{}{colName},
 	}
 

ddl/ddl_worker.go

@@ -401,19 +401,9 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 		metrics.DDLWorkerHistogram.WithLabelValues(metrics.WorkerFinishDDLJob, job.Type.String(), metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
 	}()
 
-	if !job.IsCancelled() {
-		switch job.Type {
-		case model.ActionAddIndex, model.ActionAddPrimaryKey:
-			if job.State != model.JobStateRollbackDone {
-				break
-			}
-
-			// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
-			err = w.deleteRange(w.ddlJobCtx, job)
-		case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropPrimaryKey,
-			model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionDropColumn, model.ActionDropColumns, model.ActionModifyColumn, model.ActionDropIndexes:
-			err = w.deleteRange(w.ddlJobCtx, job)
-		}
+	err = deleteRangeForDropSchemaObjectJob(w, job)
+	if err != nil {
+		return errors.Trace(err)
 	}
 
 	switch job.Type {
@@ -448,6 +438,33 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 	return errors.Trace(err)
 }
 
+func deleteRangeForDropSchemaObjectJob(w *worker, job *model.Job) error {
+	if job.IsCancelled() {
+		return nil
+	}
+	switch job.Type {
+	case model.ActionAddIndex, model.ActionAddPrimaryKey:
+		if job.State != model.JobStateRollbackDone {
+			break
+		}
+		// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
+		return w.deleteRange(w.ddlJobCtx, job)
+	case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropPrimaryKey,
+		model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionDropColumn, model.ActionDropColumns, model.ActionModifyColumn, model.ActionDropIndexes:
+		return w.deleteRange(w.ddlJobCtx, job)
+	case model.ActionMultiSchemaChange:
+		for _, sub := range job.MultiSchemaInfo.SubJobs {
+			proxyJob := cloneFromSubJob(job, sub)
+			err := deleteRangeForDropSchemaObjectJob(w, proxyJob)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
+		return nil
+	}
+	return nil
+}
+
 func (w *worker) writeDDLSeqNum(job *model.Job) {
 	w.ddlSeqNumMu.Lock()
 	w.ddlSeqNumMu.seqNum++
@@ -768,7 +785,9 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 	}
 	// The cause of this job state is that the job is cancelled by client.
 	if job.IsCancelling() {
-		return convertJob2RollbackJob(w, d, t, job)
+		if job.Type != model.ActionMultiSchemaChange {
+			return convertJob2RollbackJob(w, d, t, job)
+		}
 	}
 
 	if !job.IsRollingback() && !job.IsCancelling() {

ddl/index.go

@@ -655,10 +655,10 @@ func onDropIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		job.SchemaState = model.StateDeleteReorganization
 	case model.StateDeleteReorganization:
 		// reorganization -> absent
-		removeDependentHiddenColumns(tblInfo, indexInfo)
-		removeIndexInfo(tblInfo, indexInfo)
 		// Set column index flag.
 		dropIndexColumnFlag(tblInfo, indexInfo)
+		removeDependentHiddenColumns(tblInfo, indexInfo)
+		removeIndexInfo(tblInfo, indexInfo)
 
 		failpoint.Inject("mockExceedErrorLimit", func(val failpoint.Value) {
 			if val.(bool) {

ddl/multi_schema_change.go

@@ -59,8 +59,8 @@ func onMultiSchemaChange(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) (ve
 			}
 			proxyJob := cloneFromSubJob(job, sub)
 			ver, err = w.runDDLJob(d, t, proxyJob)
-			mergeBackToSubJob(proxyJob, sub)
 			handleRevertibleException(job, sub.State, i)
+			mergeBackToSubJob(proxyJob, sub)
 			return ver, err
 		}
 		// All the sub-jobs are non-revertible.
@@ -95,7 +95,7 @@ func isFinished(job *model.SubJob) bool {
 
 func cloneFromSubJob(job *model.Job, sub *model.SubJob) *model.Job {
 	return &model.Job{
-		ID:              0,
+		ID:              job.ID,
 		Type:            sub.Type,
 		SchemaID:        job.SchemaID,
 		TableID:         job.TableID,
@@ -135,10 +135,10 @@ func handleRevertibleException(job *model.Job, res model.JobState, idx int) {
 	if res == model.JobStateRollingback || res == model.JobStateCancelling {
 		job.State = res
 	}
-	// Flush the rollback state and cancelled state to sub-jobs.
+	// Flush the cancelling state and cancelled state to sub-jobs.
 	for i, sub := range job.MultiSchemaInfo.SubJobs {
 		if i < idx {
-			sub.State = model.JobStateRollingback
+			sub.State = model.JobStateCancelling
 		}
 		if i > idx {
 			sub.State = model.JobStateCancelled

ddl/multi_schema_change_test.go

@@ -15,10 +15,16 @@
 package ddl_test
 
 import (
+	"context"
 	"testing"
 
+	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/errno"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/testkit"
+	"github.com/pingcap/tidb/util/admin"
+	"github.com/stretchr/testify/require"
 )
 
 func TestMultiSchemaChangeAddColumns(t *testing.T) {
@@ -71,6 +77,38 @@ func TestMultiSchemaChangeAddColumns(t *testing.T) {
 	tk.MustGetErrCode("alter table t add column b int default 2, add column b int default 3", errno.ErrUnsupportedDDLOperation)
 }
 
+func TestMultiSchemaChangeAddColumnsCancelled(t *testing.T) {
+	store, dom, clean := testkit.CreateMockStoreAndDomain(t)
+	defer clean()
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("set @@global.tidb_enable_change_multi_schema = 1")
+
+	tk.MustExec("create table t (a int);")
+	tk.MustExec("insert into t values (1);")
+	var checkErr error
+	var once bool
+	hook := func(job *model.Job) {
+		if once || job.MultiSchemaInfo.SubJobs[1].SchemaState != model.StateWriteReorganization {
+			return
+		}
+		once = true
+		checkErr = kv.RunInNewTxn(context.Background(), store, false,
+			func(ctx context.Context, txn kv.Transaction) error {
+				errs, err := admin.CancelJobs(txn, []int64{job.ID})
+				if errs[0] != nil {
+					return errs[0]
+				}
+				return err
+			})
+	}
+	dom.DDL().SetHook(&ddl.TestDDLCallback{Do: dom, OnJobUpdatedExported: hook})
+	require.NoError(t, checkErr)
+	sql := "alter table t add column b int default 2, add column c int default 3, add column d int default 4;"
+	tk.MustGetErrCode(sql, errno.ErrCancelledDDLJob)
+	tk.MustQuery("select * from t;").Check(testkit.Rows("1"))
+}
+
 func TestMultiSchemaChangeDropColumns(t *testing.T) {
 	store, clean := testkit.CreateMockStore(t)
 	defer clean()
@@ -191,6 +229,12 @@ func TestMultiSchemaChangeDropIndexes(t *testing.T) {
 	tk.MustExec("create table t (a int, b int, c int, index t(a))")
 	tk.MustGetErrCode("alter table t drop index t, drop index t", errno.ErrUnsupportedDDLOperation)
 
+	tk.MustExec("create table t (id int, c1 int, c2 int, primary key(id), key i1(c1), key i2(c2));")
+	tk.MustExec("insert into t values (1, 2, 3);")
+	tk.MustExec("alter table t drop index i1, drop index i2;")
+	tk.MustGetErrCode("select * from t use index(i1);", errno.ErrKeyDoesNotExist)
+	tk.MustGetErrCode("select * from t use index(i2);", errno.ErrKeyDoesNotExist)
+
 	// Test drop index with drop column.
 	/*
 		tk.MustExec("drop table if exists t")
@@ -236,17 +280,3 @@ func TestMultiSchemaRenameIndexes(t *testing.T) {
 	tk.MustExec("create table t (a int, b int, c int, index t(a))")
 	tk.MustGetErrCode("alter table t add index t1(b), rename index t to t1", errno.ErrUnsupportedDDLOperation)
 }
-
-func TestMultiSchemaChangeDropIndexes(t *testing.T) {
-	store, clean := testkit.CreateMockStore(t)
-	defer clean()
-	tk := testkit.NewTestKit(t, store)
-	tk.MustExec("use test;")
-	tk.MustExec("set @@global.tidb_enable_change_multi_schema = 1;")
-
-	tk.MustExec("create table t (id int, c1 int, c2 int, primary key(id), key i1(c1), key i2(c2));")
-	tk.MustExec("insert into t values (1, 2, 3);")
-	tk.MustExec("alter table t drop index i1, drop index i2;")
-	tk.MustExec("select * from t use index(i1);")
-	tk.MustExec("select * from t use index(i2);")
-}

util/admin/admin.go

@@ -119,6 +119,8 @@ func IsJobRollbackable(job *model.Job) bool {
 		model.ActionModifySchemaCharsetAndCollate, model.ActionRepairTable,
 		model.ActionModifyTableAutoIdCache, model.ActionModifySchemaDefaultPlacement:
 		return job.SchemaState == model.StateNone
+	case model.ActionMultiSchemaChange:
+		return job.MultiSchemaInfo.Revertible
 	}
 	return true
 }
@@ -166,6 +168,9 @@ func CancelJobs(txn kv.Transaction, ids []int64) ([]error, error) {
 			errs[i] = ErrCannotCancelDDLJob.GenWithStackByArgs(job.ID)
 			continue
 		}
+		if job.IsCancelling() {
+			continue
+		}
 
 		job.State = model.JobStateCancelling
 		// Make sure RawArgs isn't overwritten.