ddl: Corrected index management during REORGANIZE PARTITION

pkg/ddl/partition.go

@@ -2186,8 +2186,8 @@ func (w *worker) rollbackLikeDropPartition(jobCtx *jobContext, job *model.Job) (
 	var dropIndices []*model.IndexInfo
 	for _, indexInfo := range tblInfo.Indices {
 		if indexInfo.Unique &&
-			indexInfo.State == model.StateDeleteReorganization &&
-			tblInfo.Partition.DDLState == model.StateDeleteReorganization {
+			indexInfo.State == model.StateWriteOnly &&
+			tblInfo.Partition.DDLState == model.StateWriteReorganization {
 			dropIndices = append(dropIndices, indexInfo)
 		}
 	}
@@ -3151,6 +3151,9 @@ func getReorgPartitionInfo(t *meta.Mutator, job *model.Job, args *model.TablePar
 //
 //	Everything now looks as it should, no memory of old partitions/indexes,
 //	and no more double writing, since the previous state is only reading the new partitions/indexes.
+//
+// Note: Special handling is also required in tables.newPartitionedTable(),
+// to get per partition indexes in the right state.
 func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver int64, _ error) {
 	args, err := model.GetTablePartitionArgs(job)
 	if err != nil {
@@ -3262,6 +3265,9 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 			if err != nil {
 				return ver, errors.Trace(err)
 			}
+			if tblInfo.Partition.DDLChangedIndex == nil {
+				tblInfo.Partition.DDLChangedIndex = make(map[int64]bool)
+			}
 			if !inAllPartitionColumns {
 				// Currently only support Explicit Global indexes.
 				if !newGlobal {
@@ -3279,6 +3285,8 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 				newIndex := index.Clone()
 				newIndex.State = model.StateDeleteOnly
 				newIndex.ID = AllocateIndexID(tblInfo)
+				tblInfo.Partition.DDLChangedIndex[index.ID] = false
+				tblInfo.Partition.DDLChangedIndex[newIndex.ID] = true
 				newIndex.Global = true
 				tblInfo.Indices = append(tblInfo.Indices, newIndex)
 			} else {
@@ -3291,6 +3299,8 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 					newIndex := index.Clone()
 					newIndex.State = model.StateDeleteOnly
 					newIndex.ID = AllocateIndexID(tblInfo)
+					tblInfo.Partition.DDLChangedIndex[index.ID] = false
+					tblInfo.Partition.DDLChangedIndex[newIndex.ID] = true
 					newIndex.Global = false
 					tblInfo.Indices = append(tblInfo.Indices, newIndex)
 				}
@@ -3487,26 +3497,18 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 			if !index.Unique {
 				continue
 			}
-			switch index.State {
-			case model.StateWriteReorganization:
+			isNew, ok := tblInfo.Partition.DDLChangedIndex[index.ID]
+			if !ok {
+				continue
+			}
+			if isNew {
 				// Newly created index, replacing old unique/global index
 				index.State = model.StatePublic
-			case model.StatePublic:
-				if index.Global {
-					// Mark the old global index as non-readable, and to be dropped
-					index.State = model.StateDeleteReorganization
-				} else {
-					inAllPartitionColumns, err := checkPartitionKeysConstraint(partInfo, index.Columns, tblInfo)
-					if err != nil {
-						return rollbackReorganizePartitionWithErr(jobCtx, job, err)
-					}
-					if !inAllPartitionColumns {
-						// Mark the old unique index as non-readable, and to be dropped,
-						// since it is replaced by a global index
-						index.State = model.StateDeleteReorganization
-					}
-				}
+				continue
 			}
+			// Old index, should not be visible any longer,
+			// but needs to be kept up-to-date in case rollback happens.
+			index.State = model.StateWriteOnly
 		}
 		firstPartIdx, lastPartIdx, idMap, err2 := getReplacedPartitionIDs(partNames, tblInfo.Partition)
 		if err2 != nil {
@@ -3563,14 +3565,18 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 
 		var dropIndices []*model.IndexInfo
 		for _, indexInfo := range tblInfo.Indices {
-			if indexInfo.Unique && indexInfo.State == model.StateDeleteReorganization {
+			if indexInfo.Unique && indexInfo.State == model.StateWriteOnly {
 				// Drop the old unique (possible global) index, see onDropIndex
 				indexInfo.State = model.StateNone
 				DropIndexColumnFlag(tblInfo, indexInfo)
 				RemoveDependentHiddenColumns(tblInfo, indexInfo)
 				dropIndices = append(dropIndices, indexInfo)
 			}
 		}
+		// TODO: verify that the indexes are dropped,
+		// and that StateDeleteOnly+StateDeleteReorganization is not needed.
+		// local indexes is not an issue, since they will be gone with the dropped
+		// partitions, but replaced global indexes should be checked!
 		for _, indexInfo := range dropIndices {
 			removeIndexInfo(tblInfo, indexInfo)
 		}
@@ -3632,6 +3638,9 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 				failpoint.Return(ver, errors.New("Injected error by reorgPartFail5"))
 			}
 		})
+		failpoint.Inject("updateVersionAndTableInfoErrInStateDeleteReorganization", func() {
+			failpoint.Return(ver, errors.New("Injected error in StateDeleteReorganization"))
+		})
 		args.OldPhysicalTblIDs = physicalTableIDs
 		args.NewPartitionIDs = newIDs
 		ver, err = updateVersionAndTableInfo(jobCtx, job, tblInfo, true)

pkg/ddl/rollingback.go

@@ -331,8 +331,6 @@ func convertAddTablePartitionJob2RollbackJob(jobCtx *jobContext, job *model.Job,
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
-	tblInfo.Partition.DDLState = model.StateNone
-	tblInfo.Partition.DDLAction = model.ActionNone
 	job.State = model.JobStateRollingback
 	return ver, errors.Trace(otherwiseErr)
 }
@@ -371,51 +369,25 @@ func convertReorgPartitionJob2RollbackJob(jobCtx *jobContext, job *model.Job, ot
 		partNames = append(partNames, pd.Name.L)
 	}
 	var dropIndices []*model.IndexInfo
-	// When Global Index is duplicated to a non Global, we later need
-	// to know if if it was Global before (marked to be dropped) or not.
-	globalToUniqueDupMap := make(map[string]int64)
 	for _, indexInfo := range tblInfo.Indices {
 		if !indexInfo.Unique {
 			continue
 		}
-		switch indexInfo.State {
-		case model.StateWriteReorganization, model.StateDeleteOnly,
-			model.StateWriteOnly:
-			dropIndices = append(dropIndices, indexInfo)
-		case model.StateDeleteReorganization:
-			if pi.DDLState != model.StateDeleteReorganization {
-				continue
-			}
-			// Old index marked to be dropped, rollback by making it public again
-			indexInfo.State = model.StatePublic
-			if indexInfo.Global {
-				if id, ok := globalToUniqueDupMap[indexInfo.Name.L]; ok {
-					return ver, errors.NewNoStackErrorf("Duplicate global index names '%s', %d != %d", indexInfo.Name.O, indexInfo.ID, id)
-				}
-				globalToUniqueDupMap[indexInfo.Name.L] = indexInfo.ID
-			}
-		case model.StatePublic:
-			if pi.DDLState != model.StateDeleteReorganization {
-				continue
+		isNew, ok := pi.DDLChangedIndex[indexInfo.ID]
+		if !ok {
+			// non-changed index
+			continue
+		}
+		if !isNew {
+			if pi.DDLState == model.StateDeleteReorganization {
+				// Revert the non-public state
+				indexInfo.State = model.StatePublic
 			}
-			// We cannot drop the index here, we need to wait until
-			// the next schema version
-			// i.e. rollback in rollbackLikeDropPartition
-			// New index that became public in this state,
-			// mark it to be dropped in next schema version
-			if indexInfo.Global {
-				indexInfo.State = model.StateDeleteReorganization
+		} else {
+			if pi.DDLState == model.StateDeleteReorganization {
+				indexInfo.State = model.StateWriteOnly
 			} else {
-				// How to know if this index was created as a duplicate or not?
-				if id, ok := globalToUniqueDupMap[indexInfo.Name.L]; ok {
-					// The original index
-					if id >= indexInfo.ID {
-						return ver, errors.NewNoStackErrorf("Indexes in wrong order during rollback, '%s', %d >= %d", indexInfo.Name.O, id, indexInfo.ID)
-					}
-					indexInfo.State = model.StateDeleteReorganization
-				} else {
-					globalToUniqueDupMap[indexInfo.Name.L] = indexInfo.ID
-				}
+				dropIndices = append(dropIndices, indexInfo)
 			}
 		}
 	}
@@ -466,13 +438,19 @@ func convertReorgPartitionJob2RollbackJob(jobCtx *jobContext, job *model.Job, ot
 				return ver, errors.Trace(errors.New("Internal error, failed to find original partition definitions"))
 			}
 			pi.Definitions = newDefs
-			pi.Num = uint64(len(pi.Definitions))
 		} else {
+			// Move back to StateWriteReorganization, i.e. use the original table
+			// (non-partitioned or differently partitioned) as the main table to use.
+			// Otherwise, the Type does not match the expression.
 			pi.Type, pi.DDLType = pi.DDLType, pi.Type
 			pi.Expr, pi.DDLExpr = pi.DDLExpr, pi.Expr
 			pi.Columns, pi.DDLColumns = pi.DDLColumns, pi.Columns
 			pi.Definitions = pi.DroppingDefinitions
 		}
+		pi.Num = uint64(len(pi.Definitions))
+		// We should move back one state, since there might be other sessions seeing the new partitions.
+		job.SchemaState = model.StateWriteReorganization
+		pi.DDLState = job.SchemaState
 	}
 
 	args := jobCtx.jobArgs.(*model.TablePartitionArgs)