*: Reorganize partition: tiflash and update table | tidb-test=pr/2069

ddl/BUILD.bazel

@@ -257,6 +257,7 @@ go_test(
         "//util/domainutil",
         "//util/gcutil",
         "//util/logutil",
+        "//util/mathutil",
         "//util/mock",
         "//util/sem",
         "//util/sqlexec",

ddl/backfilling.go

@@ -144,7 +144,7 @@ func GetLeaseGoTime(currTime time.Time, lease time.Duration) types.Time {
 // 1: add-index
 // 2: modify-column-type
 // 3: clean-up global index
-// 4: reorganize partition
+// 4: reorganize partition (copy data between partitions + create indexes on those new partitions)
 //
 // They all have a write reorganization state to back fill data into the rows existed.
 // Backfilling is time consuming, to accelerate this process, TiDB has built some sub
@@ -610,6 +610,7 @@ func (dc *ddlCtx) sendTasksAndWait(scheduler *backfillScheduler, totalAddedCount
 		return errors.Trace(err)
 	}
 
+	// TODO: Maybe do reorgInfo.UpdateReorgMeta here too?
 	// nextHandle will be updated periodically in runReorgJob, so no need to update it here.
 	dc.getReorgCtx(reorgInfo.Job.ID).setNextKey(nextKey)
 	metrics.BatchAddIdxHistogram.WithLabelValues(metrics.LblOK).Observe(elapsedTime.Seconds())
@@ -624,7 +625,7 @@ func (dc *ddlCtx) sendTasksAndWait(scheduler *backfillScheduler, totalAddedCount
 	return nil
 }
 
-func getBatchTasks(t table.Table, reorgInfo *reorgInfo, kvRanges []kv.KeyRange, batch int) []*reorgBackfillTask {
+func getBatchTasks(t table.PhysicalTable, reorgInfo *reorgInfo, kvRanges []kv.KeyRange, batch int) []*reorgBackfillTask {
 	batchTasks := make([]*reorgBackfillTask, 0, batch)
 	physicalTableID := reorgInfo.PhysicalTableID
 	var prefix kv.Key
@@ -654,13 +655,11 @@ func getBatchTasks(t table.Table, reorgInfo *reorgInfo, kvRanges []kv.KeyRange,
 			endKey = prefix.PrefixNext()
 		}
 
-		//nolint:forcetypeassert
-		phyTbl := t.(table.PhysicalTable)
 		task := &reorgBackfillTask{
 			id:              i,
 			jobID:           reorgInfo.Job.ID,
 			physicalTableID: physicalTableID,
-			physicalTable:   phyTbl,
+			physicalTable:   t,
 			priority:        reorgInfo.Priority,
 			startKey:        startKey,
 			endKey:          endKey,

ddl/db_integration_test.go

@@ -4382,9 +4382,7 @@ PARTITION BY LIST (level) (
 	tk.MustContainErrMsg(`ALTER TABLE members REORGANIZE PARTITION p1800,p2000 INTO (PARTITION p2000 VALUES LESS THAN (2100))`, "[ddl:8200]Unsupported REORGANIZE PARTITION of RANGE; not adjacent partitions")
 	tk.MustExec(`INSERT INTO members VALUES (313, "John", "Doe", "2022-11-22", NULL)`)
 	tk.MustExec(`ALTER TABLE members REORGANIZE PARTITION p2000 INTO (PARTITION p2000 VALUES LESS THAN (2050))`)
-	// TODO: uncomment this:
-	//tk.MustContainErrMsg(`ALTER TABLE members REORGANIZE PARTITION p2000 INTO (PARTITION p2000 VALUES LESS THAN (2020))`, "[table:1526]Table has no partition for value 2022")
+	tk.MustContainErrMsg(`ALTER TABLE members REORGANIZE PARTITION p2000 INTO (PARTITION p2000 VALUES LESS THAN (2020))`, "[table:1526]Table has no partition for value 2022")
 	tk.MustExec(`INSERT INTO member_level (id, level) values (313, 6)`)
-	// TODO: uncomment this:
-	//tk.MustContainErrMsg(`ALTER TABLE member_level REORGANIZE PARTITION lEven INTO (PARTITION lEven VALUES IN (2,4))`, "[table:1526]Table has no partition for value 6")
+	tk.MustContainErrMsg(`ALTER TABLE member_level REORGANIZE PARTITION lEven INTO (PARTITION lEven VALUES IN (2,4))`, "[table:1526]Table has no partition for value 6")
 }

ddl/ddl.go

@@ -883,7 +883,8 @@ func getIntervalFromPolicy(policy []time.Duration, i int) (time.Duration, bool)
 
 func getJobCheckInterval(job *model.Job, i int) (time.Duration, bool) {
 	switch job.Type {
-	case model.ActionAddIndex, model.ActionAddPrimaryKey, model.ActionModifyColumn:
+	case model.ActionAddIndex, model.ActionAddPrimaryKey, model.ActionModifyColumn,
+		model.ActionReorganizePartition:
 		return getIntervalFromPolicy(slowDDLIntervalPolicy, i)
 	case model.ActionCreateTable, model.ActionCreateSchema:
 		return getIntervalFromPolicy(fastDDLIntervalPolicy, i)

ddl/ddl_worker.go

@@ -40,6 +40,7 @@ import (
 	tidbutil "github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/dbterror"
 	"github.com/pingcap/tidb/util/logutil"
+	"github.com/pingcap/tidb/util/mathutil"
 	"github.com/pingcap/tidb/util/resourcegrouptag"
 	"github.com/pingcap/tidb/util/topsql"
 	topsqlstate "github.com/pingcap/tidb/util/topsql/state"
@@ -514,7 +515,8 @@ func jobNeedGC(job *model.Job) bool {
 		switch job.Type {
 		case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropPrimaryKey,
 			model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionDropColumn, model.ActionModifyColumn,
-			model.ActionAddIndex, model.ActionAddPrimaryKey:
+			model.ActionAddIndex, model.ActionAddPrimaryKey,
+			model.ActionReorganizePartition:
 			return true
 		case model.ActionMultiSchemaChange:
 			for _, sub := range job.MultiSchemaInfo.SubJobs {
@@ -973,6 +975,7 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 	// by kv reset when meets a unhandled error, but the `job` modification can't.
 	// So make sure job state and args change is after all other checks or make sure these
 	// change has no effect when retrying it.
+	// TODO: Should this ^^ comment be updated with how to handle the tidb_ddl_reorg table?
 	switch job.Type {
 	case model.ActionCreateSchema:
 		ver, err = onCreateSchema(d, t, job)
@@ -1359,6 +1362,22 @@ func updateSchemaVersion(d *ddlCtx, t *meta.Meta, job *model.Job, multiInfos ...
 				diff.AffectedOpts = buildPlacementAffects(oldIDs, oldIDs)
 			}
 		}
+	case model.ActionReorganizePartition:
+		diff.TableID = job.TableID
+		if len(job.CtxVars) > 0 {
+			if droppedIDs, ok := job.CtxVars[0].([]int64); ok {
+				if addedIDs, ok := job.CtxVars[1].([]int64); ok {
+					// to use AffectedOpts we need both new and old to have the same length
+					maxParts := mathutil.Max[int](len(droppedIDs), len(addedIDs))
+					// Also initialize them to 0!
+					oldIDs := make([]int64, maxParts)
+					copy(oldIDs, droppedIDs)
+					newIDs := make([]int64, maxParts)
+					copy(newIDs, addedIDs)
+					diff.AffectedOpts = buildPlacementAffects(oldIDs, newIDs)
+				}
+			}
+		}
 	case model.ActionCreateTable:
 		diff.TableID = job.TableID
 		if len(job.Args) > 0 {

ddl/delete_range.go

@@ -307,9 +307,13 @@ func insertJobIntoDeleteRangeTable(ctx context.Context, sctx sessionctx.Context,
 		endKey := tablecodec.EncodeTablePrefix(tableID + 1)
 		elemID := ea.allocForPhysicalID(tableID)
 		return doInsert(ctx, s, job.ID, elemID, startKey, endKey, now, fmt.Sprintf("table ID is %d", tableID))
-	case model.ActionDropTablePartition, model.ActionTruncateTablePartition:
+	case model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionReorganizePartition:
 		var physicalTableIDs []int64
-		if err := job.DecodeArgs(&physicalTableIDs); err != nil {
+		// partInfo is not used, but is set in ReorgPartition.
+		// Better to have an additional argument in job.DecodeArgs since it is ignored,
+		// instead of having one to few, which will remove the data from the job arguments...
+		var partInfo model.PartitionInfo
+		if err := job.DecodeArgs(&physicalTableIDs, &partInfo); err != nil {
 			return errors.Trace(err)
 		}
 		for _, physicalTableID := range physicalTableIDs {

ddl/index.go

@@ -1745,6 +1745,8 @@ func (w *worker) updateReorgInfo(t table.PartitionedTable, reorg *reorgInfo) (bo
 	}
 	if pid == 0 {
 		// Next partition does not exist, all the job done.
+		// TODO: Remove the ReorgMeta job?
+		// Or better, set to last key, since already processed!
 		return true, nil
 	}
 

ddl/partition.go

@@ -184,10 +184,9 @@ func (w *worker) onAddTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (v
 		if tblInfo.TiFlashReplica != nil && tblInfo.TiFlashReplica.Available {
 			// For available state, the new added partition should wait it's replica to
 			// be finished. Otherwise the query to this partition will be blocked.
-			needRetry, err := checkPartitionReplica(tblInfo.TiFlashReplica.Count, addingDefinitions, d)
+			needRetry, err := checkPartitionReplica(tblInfo.TiFlashReplica.Count, 1, addingDefinitions, d)
 			if err != nil {
-				ver, err = convertAddTablePartitionJob2RollbackJob(d, t, job, err, tblInfo)
-				return ver, err
+				return convertAddTablePartitionJob2RollbackJob(d, t, job, err, tblInfo)
 			}
 			if needRetry {
 				// The new added partition hasn't been replicated.
@@ -250,8 +249,9 @@ func alterTableLabelRule(schemaName string, meta *model.TableInfo, ids []int64)
 		if err != nil {
 			return false, errors.Wrapf(err, "failed to notify PD label rule")
 		}
+		return true, nil
 	}
-	return true, nil
+	return false, nil
 }
 
 func alterTablePartitionBundles(t *meta.Meta, tblInfo *model.TableInfo, addingDefinitions []model.PartitionDefinition) ([]*placement.Bundle, error) {
@@ -352,12 +352,17 @@ func checkAddPartitionValue(meta *model.TableInfo, part *model.PartitionInfo) er
 	return nil
 }
 
-func checkPartitionReplica(replicaCount uint64, addingDefinitions []model.PartitionDefinition, d *ddlCtx) (needWait bool, err error) {
+func checkPartitionReplica(replicaCount, minCount uint64, addingDefinitions []model.PartitionDefinition, d *ddlCtx) (needWait bool, err error) {
 	failpoint.Inject("mockWaitTiFlashReplica", func(val failpoint.Value) {
 		if val.(bool) {
 			failpoint.Return(true, nil)
 		}
 	})
+	failpoint.Inject("mockWaitTiFlashReplicaOK", func(val failpoint.Value) {
+		if val.(bool) {
+			failpoint.Return(false, nil)
+		}
+	})
 
 	ctx := context.Background()
 	pdCli := d.store.(tikv.Storage).GetRegionCache().PDClient()
@@ -388,29 +393,33 @@ func checkPartitionReplica(replicaCount uint64, addingDefinitions []model.Partit
 			if err != nil {
 				return needWait, errors.Trace(err)
 			}
-			tiflashPeerAtLeastOne := checkTiFlashPeerStoreAtLeastOne(stores, regionState.Meta.Peers)
+			tiflashPeerOK := checkTiFlashPeerStore(stores, minCount, regionState.Meta.Peers)
 			failpoint.Inject("ForceTiflashNotAvailable", func(v failpoint.Value) {
-				tiflashPeerAtLeastOne = v.(bool)
+				tiflashPeerOK = v.(bool)
 			})
 			// It's unnecessary to wait all tiflash peer to be replicated.
 			// Here only make sure that tiflash peer count > 0 (at least one).
-			if tiflashPeerAtLeastOne {
+			if tiflashPeerOK {
 				continue
 			}
 			needWait = true
-			logutil.BgLogger().Info("[ddl] partition replicas check failed in replica-only DDL state", zap.Int64("pID", pd.ID), zap.Uint64("wait region ID", region.Meta.Id), zap.Bool("tiflash peer at least one", tiflashPeerAtLeastOne), zap.Time("check time", time.Now()))
+			logutil.BgLogger().Info("[ddl] partition replicas check failed in replica-only DDL state", zap.Int64("pID", pd.ID), zap.Uint64("wait region ID", region.Meta.Id), zap.Bool("tiflash peer count OK", tiflashPeerOK), zap.Time("check time", time.Now()))
 			return needWait, nil
 		}
 	}
 	logutil.BgLogger().Info("[ddl] partition replicas check ok in replica-only DDL state")
 	return needWait, nil
 }
 
-func checkTiFlashPeerStoreAtLeastOne(stores []*metapb.Store, peers []*metapb.Peer) bool {
+func checkTiFlashPeerStore(stores []*metapb.Store, minCount uint64, peers []*metapb.Peer) bool {
+	cnt := uint64(0)
 	for _, peer := range peers {
 		for _, store := range stores {
 			if peer.StoreId == store.Id && storeHasEngineTiFlashLabel(store) {
-				return true
+				cnt++
+				if cnt >= minCount {
+					return true
+				}
 			}
 		}
 	}
@@ -1705,11 +1714,12 @@ func (w *worker) onDropTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
-	if job.Type == model.ActionAddTablePartition {
+	if job.Type == model.ActionAddTablePartition || job.Type == model.ActionReorganizePartition {
 		// It is rollbacked from adding table partition, just remove addingDefinitions from tableInfo.
 		physicalTableIDs, pNames, rollbackBundles := rollbackAddingPartitionInfo(tblInfo)
 		err = infosync.PutRuleBundlesWithDefaultRetry(context.TODO(), rollbackBundles)
 		if err != nil {
+			// Why cancel here and not try to clean up as much as possible?
 			job.State = model.JobStateCancelled
 			return ver, errors.Wrapf(err, "failed to notify PD the placement rules")
 		}
@@ -1813,6 +1823,22 @@ func (w *worker) onDropTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (
 				return ver, errors.Trace(err)
 			}
 		}
+		if tblInfo.TiFlashReplica != nil {
+			// Remove the partitions
+			ids := tblInfo.TiFlashReplica.AvailablePartitionIDs
+			// Rarely called, so OK to take some time, to make it easy
+			for _, id := range physicalTableIDs {
+				for i, avail := range ids {
+					if id == avail {
+						tmp := ids[:i]
+						tmp = append(tmp, ids[i+1:]...)
+						ids = tmp
+						break
+					}
+				}
+			}
+			tblInfo.TiFlashReplica.AvailablePartitionIDs = ids
+		}
 		tblInfo.Partition.DroppingDefinitions = nil
 		// used by ApplyDiff in updateSchemaVersion
 		job.CtxVars = []interface{}{physicalTableIDs}
@@ -2373,20 +2399,25 @@ func (w *worker) onReorganizePartition(d *ddlCtx, t *meta.Meta, job *model.Job)
 			// For available state, the new added partition should wait its replica to
 			// be finished, otherwise the query to this partition will be blocked.
 			count := tblInfo.TiFlashReplica.Count
-			needRetry, err := checkPartitionReplica(count, addingDefinitions, d)
+			needRetry, err := checkPartitionReplica(count, count, addingDefinitions, d)
 			if err != nil {
 				// need to rollback, since we tried to register the new
 				// partitions before!
 				return convertAddTablePartitionJob2RollbackJob(d, t, job, err, tblInfo)
 			}
+			// Try for 10 rounds (in case of transient TiFlash issues)
 			if needRetry {
-				// The new added partition hasn't been replicated.
-				// Do nothing to the job this time, wait next worker round.
-				time.Sleep(tiflashCheckTiDBHTTPAPIHalfInterval)
-				// Set the error here which will lead this job exit when it's retry times beyond the limitation.
-				return ver, errors.Errorf("[ddl] add partition wait for tiflash replica to complete")
-			}
+				if job.ErrorCount < 10 {
+					// The new added partition hasn't been replicated.
+					// Do nothing to the job this time, wait next worker round.
+					time.Sleep(tiflashCheckTiDBHTTPAPIHalfInterval)
+					// Set the error here which will lead this job exit when it's retry times beyond the limitation.
+					return ver, errors.Errorf("[ddl] reorganize partition wait for tiflash replica to complete")
+				}
 
+				logutil.Logger(w.logCtx).Error("[ddl] reorganize partition could not find all new regions in TiFlash replicas, skipping syncing to TiFlash!", zap.String("table", tblInfo.Name.O))
+				return convertAddTablePartitionJob2RollbackJob(d, t, job, err, tblInfo)
+			}
 			// When TiFlash Replica is ready, we must move them into `AvailablePartitionIDs`.
 			// Since onUpdateFlashReplicaStatus cannot see the partitions yet (not public)
 			for _, d := range addingDefinitions {
@@ -2785,6 +2816,7 @@ func (w *worker) reorgPartitionDataAndIndex(t table.Table, reorgInfo *reorgInfo)
 		// First run, have not yet started backfilling index data
 		// Restart with the first new partition.
 		// TODO: handle remove partitioning
+		// TODO: Check if there is a better way to restart?
 		reorgInfo.PhysicalTableID = firstNewPartitionID
 	} else {
 		// The job was interrupted and has been restarted,

ddl/reorg.go

@@ -18,6 +18,7 @@ import (
 	"encoding/hex"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -300,6 +301,11 @@ func updateBackfillProgress(w *worker, reorgInfo *reorgInfo, tblInfo *model.Tabl
 		if progress > 1 {
 			progress = 1
 		}
+		// TODO: change back to Debug
+		logutil.BgLogger().Info("[ddl] update progress",
+			zap.Float64("progress", progress),
+			zap.Int64("addedRowCount", addedRowCount),
+			zap.Int64("totalCount", totalCount))
 	}
 	switch reorgInfo.Type {
 	case model.ActionAddIndex, model.ActionAddPrimaryKey:
@@ -330,9 +336,20 @@ func getTableTotalCount(w *worker, tblInfo *model.TableInfo) int64 {
 	if !ok {
 		return statistics.PseudoRowCount
 	}
-	// TODO: if Reorganize Partition, only select number of rows from the selected partitions!
-	sql := "select table_rows from information_schema.tables where tidb_table_id=%?;"
-	rows, _, err := executor.ExecRestrictedSQL(w.ctx, nil, sql, tblInfo.ID)
+	var rows []chunk.Row
+	if tblInfo.Partition != nil && len(tblInfo.Partition.DroppingDefinitions) > 0 {
+		// if Reorganize Partition, only select number of rows from the selected partitions!
+		defs := tblInfo.Partition.DroppingDefinitions
+		partIDs := make([]string, 0, len(defs))
+		for _, def := range defs {
+			partIDs = append(partIDs, strconv.FormatInt(def.ID, 10))
+		}
+		sql := "select sum(table_rows) from information_schema.partitions where tidb_partition_id in (%?);"
+		rows, _, err = executor.ExecRestrictedSQL(w.ctx, nil, sql, strings.Join(partIDs, ","))
+	} else {
+		sql := "select table_rows from information_schema.tables where tidb_table_id=%?;"
+		rows, _, err = executor.ExecRestrictedSQL(w.ctx, nil, sql, tblInfo.ID)
+	}
 	if err != nil {
 		return statistics.PseudoRowCount
 	}
@@ -553,12 +570,12 @@ func getTableRange(ctx *JobContext, d *ddlCtx, tbl table.PhysicalTable, snapshot
 		endHandleKey = tablecodec.EncodeRecordKey(tbl.RecordPrefix(), maxHandle)
 	}
 	if isEmptyTable || endHandleKey.Cmp(startHandleKey) < 0 {
-		logutil.BgLogger().Info("[ddl] get noop table range",
+		logutil.BgLogger().Info("[ddl] get table range, endHandle < startHandle",
 			zap.String("table", fmt.Sprintf("%v", tbl.Meta())),
 			zap.Int64("table/partition ID", tbl.GetPhysicalID()),
-			zap.String("start key", hex.EncodeToString(startHandleKey)),
-			zap.String("end key", hex.EncodeToString(endHandleKey)),
-			zap.Bool("is empty table", isEmptyTable))
+			zap.Bool("isEmptyTable", isEmptyTable),
+			zap.String("endHandle", hex.EncodeToString(endHandleKey)),
+			zap.String("startHandle", hex.EncodeToString(startHandleKey)))
 		endHandleKey = startHandleKey
 	}
 	return
@@ -694,7 +711,6 @@ func getReorgInfoFromPartitions(ctx *JobContext, d *ddlCtx, rh *reorgHandler, jo
 		}
 		pid = partitionIDs[0]
 		physTbl := tbl.GetPartition(pid)
-
 		start, end, err = getTableRange(ctx, d, physTbl, ver.Ver, job.Priority)
 		if err != nil {
 			return nil, errors.Trace(err)

ddl/sanity_check.go

@@ -98,7 +98,8 @@ func expectedDeleteRangeCnt(ctx delRangeCntCtx, job *model.Job) (int, error) {
 			return 0, errors.Trace(err)
 		}
 		return mathutil.Max(len(physicalTableIDs), 1), nil
-	case model.ActionDropTablePartition, model.ActionTruncateTablePartition:
+	case model.ActionDropTablePartition, model.ActionTruncateTablePartition,
+		model.ActionReorganizePartition:
 		var physicalTableIDs []int64
 		if err := job.DecodeArgs(&physicalTableIDs); err != nil {
 			return 0, errors.Trace(err)

ddl/tiflashtest/BUILD.bazel

@@ -22,6 +22,7 @@ go_test(
         "//store/mockstore/unistore",
         "//table",
         "//testkit",
+        "//testkit/external",
         "//testkit/testsetup",
         "//util",
         "//util/logutil",