ddl: Support the operation of dropping multi-indexes in one statement

ddl/ddl_api.go

@@ -2278,6 +2278,8 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 				err = d.AddColumns(ctx, ident, validSpecs)
 			case ast.AlterTableDropColumn:
 				err = d.DropColumns(ctx, ident, validSpecs)
+			case ast.AlterTableDropIndex:
+				err = d.DropIndexes(ctx, ident, validSpecs)
 			default:
 				return errRunMultiSchemaChanges
 			}
@@ -4789,7 +4791,6 @@ func (d *ddl) CreateForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName mode
 	err = d.doDDLJob(ctx, job)
 	err = d.callHookOnChanged(err)
 	return errors.Trace(err)
-
 }
 
 func (d *ddl) DropForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName model.CIStr) error {
@@ -4891,6 +4892,93 @@ func (d *ddl) DropIndex(ctx sessionctx.Context, ti ast.Ident, indexName model.CI
 	return errors.Trace(err)
 }
 
+func (d *ddl) DropIndexes(ctx sessionctx.Context, ti ast.Ident, specs []*ast.AlterTableSpec) error {
+	is := d.infoHandle.Get()
+	schema, ok := is.SchemaByName(ti.Schema)
+	if !ok {
+		return errors.Trace(infoschema.ErrDatabaseNotExists)
+	}
+	t, err := is.TableByName(ti.Schema, ti.Name)
+	if err != nil {
+		return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ti.Schema, ti.Name))
+	}
+
+	allIndexIsExists := true
+	indexNames := make([]interface{}, 0, len(specs))
+	for _, spec := range specs {
+		var indexName model.CIStr
+		switch spec.Tp {
+		case ast.AlterTableDropIndex:
+			indexName = model.NewCIStr(spec.Name)
+		case ast.AlterTableDropPrimaryKey:
+			indexName = model.NewCIStr(mysql.PrimaryKeyName)
+		default:
+			return nil
+		}
+
+		indexInfo := t.Meta().FindIndexByName(indexName.L)
+		var isPK bool
+		if indexName.L == strings.ToLower(mysql.PrimaryKeyName) &&
+			// Before we fixed #14243, there might be a general index named `primary` but not a primary key.
+			(indexInfo == nil || indexInfo.Primary) {
+			isPK = true
+		}
+		if isPK {
+			if !config.GetGlobalConfig().AlterPrimaryKey {
+				return ErrUnsupportedModifyPrimaryKey.GenWithStack("Unsupported drop primary key when alter-primary-key is false")
+			}
+			// If the table's PKIsHandle is true, we can't find the index from the table. So we check the value of PKIsHandle.
+			if indexInfo == nil && !t.Meta().PKIsHandle {
+				return ErrCantDropFieldOrKey.GenWithStack("Can't DROP 'PRIMARY'; check that column/key exists")
+			}
+			if t.Meta().PKIsHandle {
+				return ErrUnsupportedModifyPrimaryKey.GenWithStack("Unsupported drop primary key when the table's pkIsHandle is true")
+			}
+			if t.Meta().IsCommonHandle {
+				return ErrUnsupportedModifyPrimaryKey.GenWithStack("Unsupported drop primary key when the table is using clustered index")
+			}
+		}
+		if indexInfo == nil {
+			err = ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName)
+			if spec.IfExists {
+				ctx.GetSessionVars().StmtCtx.AppendNote(err)
+				continue
+			}
+			return err
+		}
+
+		// Check for drop index on auto_increment column.
+		err = checkDropIndexOnAutoIncrementColumn(t.Meta(), indexInfo)
+		if err != nil {
+			return errors.Trace(err)
+		}
+
+		indexNames = append(indexNames, indexName)
+
+		if !spec.IfExists {
+			allIndexIsExists = false
+		}
+	}
+
+	job := &model.Job{
+		SchemaID:   schema.ID,
+		TableID:    t.Meta().ID,
+		SchemaName: schema.Name.L,
+		Type:       model.ActionDropIndex,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{indexNames},
+	}
+
+	err = d.doDDLJob(ctx, job)
+	// index not exists, but if_exists flags is true, so we ignore this error.
+	if ErrCantDropFieldOrKey.Equal(err) && allIndexIsExists {
+		ctx.GetSessionVars().StmtCtx.AppendNote(err)
+		return nil
+	}
+	err = d.callHookOnChanged(err)
+	return errors.Trace(err)
+}
+
 func isDroppableColumn(tblInfo *model.TableInfo, colName model.CIStr) error {
 	// Check whether there are other columns depend on this column or not.
 	for _, col := range tblInfo.Columns {

ddl/ddl_worker.go

@@ -644,7 +644,7 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 	case model.ActionAddPrimaryKey:
 		ver, err = w.onCreateIndex(d, t, job, true)
 	case model.ActionDropIndex, model.ActionDropPrimaryKey:
-		ver, err = onDropIndex(t, job)
+		ver, err = onDropIndexes(t, job)
 	case model.ActionRenameIndex:
 		ver, err = onRenameIndex(t, job)
 	case model.ActionAddForeignKey:

ddl/delete_range.go

@@ -303,44 +303,53 @@ func insertJobIntoDeleteRangeTable(ctx sessionctx.Context, job *model.Job) error
 	// ActionAddIndex, ActionAddPrimaryKey needs do it, because it needs to be rolled back when it's canceled.
 	case model.ActionAddIndex, model.ActionAddPrimaryKey:
 		tableID := job.TableID
-		var indexID int64
+		var indexIDs []int64
 		var partitionIDs []int64
-		if err := job.DecodeArgs(&indexID, &partitionIDs); err != nil {
+		if err := job.DecodeArgs(&indexIDs, &partitionIDs); err != nil {
 			return errors.Trace(err)
 		}
-		if len(partitionIDs) > 0 {
-			for _, pid := range partitionIDs {
-				startKey := tablecodec.EncodeTableIndexPrefix(pid, indexID)
-				endKey := tablecodec.EncodeTableIndexPrefix(pid, indexID+1)
-				if err := doInsert(s, job.ID, indexID, startKey, endKey, now); err != nil {
-					return errors.Trace(err)
+		for _, indexID := range indexIDs {
+			if len(partitionIDs) > 0 {
+				for _, pid := range partitionIDs {
+					startKey := tablecodec.EncodeTableIndexPrefix(pid, indexID)
+					endKey := tablecodec.EncodeTableIndexPrefix(pid, indexID+1)
+					if err := doInsert(s, job.ID, indexID, startKey, endKey, now); err != nil {
+						return errors.Trace(err)
+					}
 				}
+			} else {
+				startKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID)
+				endKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID+1)
+				return doInsert(s, job.ID, indexID, startKey, endKey, now)
 			}
-		} else {
-			startKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID)
-			endKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID+1)
-			return doInsert(s, job.ID, indexID, startKey, endKey, now)
 		}
 	case model.ActionDropIndex, model.ActionDropPrimaryKey:
 		tableID := job.TableID
-		var indexName interface{}
-		var indexID int64
+		var indexNames []model.CIStr
+		var indexIDs []int64
 		var partitionIDs []int64
-		if err := job.DecodeArgs(&indexName, &indexID, &partitionIDs); err != nil {
-			return errors.Trace(err)
+		if err := job.DecodeArgs(&indexNames, &indexIDs, &partitionIDs); err != nil {
+			var indexName model.CIStr
+			if err := job.DecodeArgs(&indexName, &indexIDs, &partitionIDs); err != nil {
+				return errors.Trace(err)
+			}
+			indexNames = append(indexNames, indexName)
 		}
-		if len(partitionIDs) > 0 {
-			for _, pid := range partitionIDs {
-				startKey := tablecodec.EncodeTableIndexPrefix(pid, indexID)
-				endKey := tablecodec.EncodeTableIndexPrefix(pid, indexID+1)
-				if err := doInsert(s, job.ID, indexID, startKey, endKey, now); err != nil {
-					return errors.Trace(err)
+
+		for _, indexID := range indexIDs {
+			if len(partitionIDs) > 0 {
+				for _, pid := range partitionIDs {
+					startKey := tablecodec.EncodeTableIndexPrefix(pid, indexID)
+					endKey := tablecodec.EncodeTableIndexPrefix(pid, indexID+1)
+					if err := doInsert(s, job.ID, indexID, startKey, endKey, now); err != nil {
+						return errors.Trace(err)
+					}
 				}
+			} else {
+				startKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID)
+				endKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID+1)
+				return doInsert(s, job.ID, indexID, startKey, endKey, now)
 			}
-		} else {
-			startKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID)
-			endKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID+1)
-			return doInsert(s, job.ID, indexID, startKey, endKey, now)
 		}
 	case model.ActionDropColumn:
 		var colName model.CIStr

ddl/index.go

@@ -385,7 +385,7 @@ func updateHiddenColumns(tblInfo *model.TableInfo, idxInfo *model.IndexInfo, sta
 func (w *worker) onCreateIndex(d *ddlCtx, t *meta.Meta, job *model.Job, isPK bool) (ver int64, err error) {
 	// Handle the rolling back job.
 	if job.IsRollingback() {
-		ver, err = onDropIndex(t, job)
+		ver, err = onDropIndexes(t, job)
 		if err != nil {
 			return ver, errors.Trace(err)
 		}
@@ -599,25 +599,34 @@ func (w *worker) onCreateIndex(d *ddlCtx, t *meta.Meta, job *model.Job, isPK boo
 	return ver, errors.Trace(err)
 }
 
-func onDropIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) {
-	tblInfo, indexInfo, err := checkDropIndex(t, job)
+func onDropIndexes(t *meta.Meta, job *model.Job) (ver int64, _ error) {
+	tblInfo, indexesInfo, err := checkDropIndexes(t, job)
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
+	if len(indexesInfo) == 0 {
+		job.State = model.JobStateCancelled
+		return ver, nil
+	}
 
 	dependentHiddenCols := make([]*model.ColumnInfo, 0)
-	for _, indexColumn := range indexInfo.Columns {
-		if tblInfo.Columns[indexColumn.Offset].Hidden {
-			dependentHiddenCols = append(dependentHiddenCols, tblInfo.Columns[indexColumn.Offset])
+	for _, indexInfo := range indexesInfo {
+		for _, indexColumn := range indexInfo.Columns {
+			if tblInfo.Columns[indexColumn.Offset].Hidden {
+				dependentHiddenCols = append(dependentHiddenCols, tblInfo.Columns[indexColumn.Offset])
+			}
 		}
 	}
 
+	indexInfo := indexesInfo[0]
 	originalState := indexInfo.State
 	switch indexInfo.State {
 	case model.StatePublic:
 		// public -> write only
 		job.SchemaState = model.StateWriteOnly
-		indexInfo.State = model.StateWriteOnly
+		for _, indexInfo := range indexesInfo {
+			indexInfo.State = model.StateWriteOnly
+		}
 		if len(dependentHiddenCols) > 0 {
 			firstHiddenOffset := dependentHiddenCols[0].Offset
 			for i := 0; i < len(dependentHiddenCols); i++ {
@@ -630,26 +639,37 @@ func onDropIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	case model.StateWriteOnly:
 		// write only -> delete only
 		job.SchemaState = model.StateDeleteOnly
-		indexInfo.State = model.StateDeleteOnly
-		updateHiddenColumns(tblInfo, indexInfo, model.StateDeleteOnly)
+		for _, indexInfo := range indexesInfo {
+			indexInfo.State = model.StateDeleteOnly
+			updateHiddenColumns(tblInfo, indexInfo, model.StateDeleteOnly)
+		}
 		ver, err = updateVersionAndTableInfo(t, job, tblInfo, originalState != indexInfo.State)
 	case model.StateDeleteOnly:
 		// delete only -> reorganization
 		job.SchemaState = model.StateDeleteReorganization
-		indexInfo.State = model.StateDeleteReorganization
-		updateHiddenColumns(tblInfo, indexInfo, model.StateDeleteReorganization)
+		for _, indexInfo := range indexesInfo {
+			indexInfo.State = model.StateDeleteReorganization
+			updateHiddenColumns(tblInfo, indexInfo, model.StateDeleteReorganization)
+		}
 		ver, err = updateVersionAndTableInfo(t, job, tblInfo, originalState != indexInfo.State)
 	case model.StateDeleteReorganization:
-		// reorganization -> absent
+		indexesIDs := make([]int64, 0, len(indexesInfo))
+		indexesLName := make(map[string]bool, len(indexesInfo))
+		for _, indexInfo := range indexesInfo {
+			indexesLName[indexInfo.Name.L] = true
+			indexesIDs = append(indexesIDs, indexInfo.ID)
+		}
 		newIndices := make([]*model.IndexInfo, 0, len(tblInfo.Indices))
 		for _, idx := range tblInfo.Indices {
-			if idx.Name.L != indexInfo.Name.L {
+			if _, ok := indexesLName[idx.Name.L]; !ok {
 				newIndices = append(newIndices, idx)
 			}
 		}
 		tblInfo.Indices = newIndices
 		// Set column index flag.
-		dropIndexColumnFlag(tblInfo, indexInfo)
+		for _, indexInfo := range indexesInfo {
+			dropIndexColumnFlag(tblInfo, indexInfo)
+		}
 
 		tblInfo.Columns = tblInfo.Columns[:len(tblInfo.Columns)-len(dependentHiddenCols)]
 
@@ -661,52 +681,65 @@ func onDropIndex(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		// Finish this job.
 		if job.IsRollingback() {
 			job.FinishTableJob(model.JobStateRollbackDone, model.StateNone, ver, tblInfo)
-			job.Args[0] = indexInfo.ID
+			job.Args[0] = indexesIDs
 			// the partition ids were append by convertAddIdxJob2RollbackJob, it is weird, but for the compatibility,
 			// we should keep appending the partitions in the convertAddIdxJob2RollbackJob.
 		} else {
 			job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
-			job.Args = append(job.Args, indexInfo.ID, getPartitionIDs(tblInfo))
+			job.Args = append(job.Args, indexesIDs, getPartitionIDs(tblInfo))
 		}
 	default:
 		err = ErrInvalidDDLState.GenWithStackByArgs("index", indexInfo.State)
 	}
 	return ver, errors.Trace(err)
 }
 
-func checkDropIndex(t *meta.Meta, job *model.Job) (*model.TableInfo, *model.IndexInfo, error) {
+func checkDropIndexes(t *meta.Meta, job *model.Job) (*model.TableInfo, []*model.IndexInfo, error) {
 	schemaID := job.SchemaID
 	tblInfo, err := getTableInfoAndCancelFaultJob(t, job, schemaID)
 	if err != nil {
 		return nil, nil, errors.Trace(err)
 	}
 
-	var indexName model.CIStr
-	if err = job.DecodeArgs(&indexName); err != nil {
-		job.State = model.JobStateCancelled
-		return nil, nil, errors.Trace(err)
+	var indexNames []model.CIStr
+	if err = job.DecodeArgs(&indexNames); err != nil {
+		var indexName model.CIStr
+		if err = job.DecodeArgs(&indexName); err != nil {
+			job.State = model.JobStateCancelled
+			return nil, nil, errors.Trace(err)
+		}
+		indexNames = append(indexNames, indexName)
 	}
 
-	indexInfo := tblInfo.FindIndexByName(indexName.L)
-	if indexInfo == nil {
-		job.State = model.JobStateCancelled
-		return nil, nil, ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName)
-	}
+	indexesInfo := make([]*model.IndexInfo, 0, len(indexNames))
+	indexesMap := make(map[string]bool)
+	for _, indexName := range indexNames {
+		indexInfo := tblInfo.FindIndexByName(indexName.L)
+		if indexInfo == nil {
+			job.State = model.JobStateCancelled
+			return nil, nil, ErrCantDropFieldOrKey.GenWithStack("index %s doesn't exist", indexName)
+		}
 
-	// Double check for drop index on auto_increment column.
-	err = checkDropIndexOnAutoIncrementColumn(tblInfo, indexInfo)
-	if err != nil {
-		job.State = model.JobStateCancelled
-		return nil, nil, autoid.ErrWrongAutoKey
+		// Double check for drop index on auto_increment column.
+		err = checkDropIndexOnAutoIncrementColumn(tblInfo, indexInfo)
+		if err != nil {
+			job.State = model.JobStateCancelled
+			return nil, nil, autoid.ErrWrongAutoKey
+		}
+
+		indexesInfo = append(indexesInfo, indexInfo)
+		indexesMap[indexName.L] = true
 	}
 
 	// Check that drop primary index will not cause invisible implicit primary index.
 	newIndices := make([]*model.IndexInfo, 0, len(tblInfo.Indices))
+
 	for _, idx := range tblInfo.Indices {
-		if idx.Name.L != indexInfo.Name.L {
+		if _, ok := indexesMap[idx.Name.L]; !ok {
 			newIndices = append(newIndices, idx)
 		}
 	}
+
 	newTbl := tblInfo.Clone()
 	newTbl.Indices = newIndices
 	err = checkInvisibleIndexOnPK(newTbl)
@@ -715,7 +748,7 @@ func checkDropIndex(t *meta.Meta, job *model.Job) (*model.TableInfo, *model.Inde
 		return nil, nil, errors.Trace(err)
 	}
 
-	return tblInfo, indexInfo, nil
+	return tblInfo, indexesInfo, nil
 }
 
 func checkDropIndexOnAutoIncrementColumn(tblInfo *model.TableInfo, indexInfo *model.IndexInfo) error {