ddl: dynamic adjust add index worker number.

ddl/callback.go

@@ -45,6 +45,8 @@ type Callback interface {
 	OnJobUpdated(job *model.Job)
 	// OnWatched is called after watching owner is completed.
 	OnWatched(ctx context.Context)
+	// OnIndexWorkerReorgBefore is called before add index worker running reorganization.
+	OnIndexWorkerReorgBefore(workerNum, rangesNum int)
 }
 
 // BaseCallback implements Callback.OnChanged interface.
@@ -70,3 +72,8 @@ func (c *BaseCallback) OnJobUpdated(job *model.Job) {
 func (c *BaseCallback) OnWatched(ctx context.Context) {
 	// Nothing to do.
 }
+
+// OnIndexWorkerReorgBefore implements Callback.OnIndexWorkerReorgBefore interface.
+func (c *BaseCallback) OnIndexWorkerReorgBefore(workerNum, rangesNum int) {
+	// Nothing to do.
+}

ddl/callback_test.go

@@ -40,11 +40,12 @@ func (ti *TestInterceptor) OnGetInfoSchema(ctx sessionctx.Context, is infoschema
 type TestDDLCallback struct {
 	*BaseCallback
 
-	onJobRunBefore         func(*model.Job)
-	OnJobRunBeforeExported func(*model.Job)
-	onJobUpdated           func(*model.Job)
-	OnJobUpdatedExported   func(*model.Job)
-	onWatched              func(ctx context.Context)
+	onJobRunBefore                   func(*model.Job)
+	OnJobRunBeforeExported           func(*model.Job)
+	onJobUpdated                     func(*model.Job)
+	OnJobUpdatedExported             func(*model.Job)
+	onWatched                        func(ctx context.Context)
+	OnIndexWorkerReorgBeforeExported func(workerNum, rangesNum int)
 }
 
 func (tc *TestDDLCallback) OnJobRunBefore(job *model.Job) {
@@ -84,6 +85,15 @@ func (tc *TestDDLCallback) OnWatched(ctx context.Context) {
 	tc.BaseCallback.OnWatched(ctx)
 }
 
+func (tc *TestDDLCallback) OnIndexWorkerReorgBefore(workerNum, rangesNum int) {
+	if tc.OnIndexWorkerReorgBeforeExported != nil {
+		tc.OnIndexWorkerReorgBeforeExported(workerNum, rangesNum)
+		return
+	}
+
+	tc.BaseCallback.OnIndexWorkerReorgBefore(workerNum, rangesNum)
+}
+
 func (s *testDDLSuite) TestCallback(c *C) {
 	cb := &BaseCallback{}
 	c.Assert(cb.OnChanged(nil), IsNil)

ddl/db_test.go

@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tidb/meta/autoid"
 	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tidb/store/mockstore/mocktikv"
 	"github.com/pingcap/tidb/table"
@@ -656,16 +657,64 @@ func (s *testDBSuite) testAddIndex(c *C, testPartition bool, createTableSQL stri
 	s.mustExec(c, sql)
 	otherKeys = append(otherKeys, v)
 
+	is := s.dom.InfoSchema()
+	schemaName := model.NewCIStr(s.schemaName)
+	tableName := model.NewCIStr("test_add_index")
+	tbl, err := is.TableByName(schemaName, tableName)
+	c.Assert(err, IsNil)
+
+	splitCount := 100
+	if !testPartition {
+		// Split table to multi region.
+		s.cluster.SplitTable(s.mvccStore, tbl.Meta().ID, splitCount)
+	}
+
+	// set hook for check add index worker num
+	hook := &ddl.TestDDLCallback{}
+	oringDDLAddIndexWorkerCnt := variable.GetDDLReorgWorkerCounter()
+	lastSetWorkerCnt := int(oringDDLAddIndexWorkerCnt)
+	defer variable.SetDDLReorgWorkerCounter(oringDDLAddIndexWorkerCnt)
+	// firstCheck is use to check split table range is take effect.
+	firstCheck := !testPartition
+	var checkErr error
+	changeWorkerNumEnable := false
+	hook.OnIndexWorkerReorgBeforeExported = func(workerNum, rangesNum int) {
+		if checkErr != nil {
+			return
+		}
+		// Check split table range is successful.
+		if firstCheck {
+			firstCheck = false
+			if rangesNum != splitCount {
+				checkErr = errors.Errorf("first check rangeNum, expect: %v, but got: %v", 100, rangesNum)
+			}
+		}
+		setNum := int(variable.GetDDLReorgWorkerCounter())
+		if rangesNum < setNum {
+			if workerNum != rangesNum {
+				checkErr = errors.Errorf("rangeNum is %v, expect workerNum is: %v, but got: %v", rangesNum, rangesNum, workerNum)
+			}
+		} else if workerNum != setNum {
+			checkErr = errors.Errorf("rangeNum is %v, expect workerNum is: %v, but got: %v", rangesNum, setNum, workerNum)
+		}
+		changeWorkerNumEnable = true
+	}
+	s.dom.DDL().(ddl.DDLForTest).SetHook(hook)
+	defer s.dom.DDL().(ddl.DDLForTest).SetHook(&ddl.TestDDLCallback{})
+
 	sessionExecInGoroutine(c, s.store, "create index c3_index on test_add_index (c3)", done)
 
 	deletedKeys := make(map[int]struct{})
 
 	ticker := time.NewTicker(s.lease / 2)
 	defer ticker.Stop()
+	ticker2 := time.NewTicker(10 * time.Millisecond)
+	defer ticker2.Stop()
+
 LOOP:
 	for {
 		select {
-		case err := <-done:
+		case err = <-done:
 			if err == nil {
 				break LOOP
 			}
@@ -688,8 +737,16 @@ LOOP:
 				s.mustExec(c, sql)
 			}
 			num += step
+		case <-ticker2.C:
+			if changeWorkerNumEnable {
+				lastSetWorkerCnt = rand.Intn(8) + 8
+				s.mustExec(c, fmt.Sprintf("set @@tidb_ddl_reorg_worker_cnt=%d", lastSetWorkerCnt))
+				c.Assert(checkErr, IsNil)
+				changeWorkerNumEnable = false
+			}
 		}
 	}
+	c.Assert(checkErr, IsNil)
 
 	// get exists keys
 	keys := make([]int, 0, num)
@@ -733,7 +790,7 @@ LOOP:
 	t := s.testGetTable(c, "test_add_index")
 	handles := make(map[int64]struct{})
 	startKey := t.RecordKey(math.MinInt64)
-	err := t.IterRecords(ctx, startKey, t.Cols(),
+	err = t.IterRecords(ctx, startKey, t.Cols(),
 		func(h int64, data []types.Datum, cols []*table.Column) (bool, error) {
 			handles[h] = struct{}{}
 			return true, nil
@@ -772,7 +829,6 @@ LOOP:
 		delete(handles, h)
 	}
 	c.Assert(handles, HasLen, 0)
-
 	s.tk.MustExec("drop table test_add_index")
 }
 

ddl/index.go

@@ -1037,18 +1037,56 @@ func (w *worker) sendRangeTaskToWorkers(t table.Table, workers []*addIndexWorker
 
 // buildIndexForReorgInfo build backfilling tasks from [reorgInfo.StartHandle, reorgInfo.EndHandle),
 // and send these tasks to add index workers, till we finish adding the indices.
-func (w *worker) buildIndexForReorgInfo(t table.PhysicalTable, workers []*addIndexWorker, job *model.Job, reorgInfo *reorgInfo) error {
+func (w *worker) buildIndexForReorgInfo(t table.PhysicalTable, indexInfo *model.IndexInfo, job *model.Job, reorgInfo *reorgInfo) error {
 	totalAddedCount := job.GetRowCount()
 
 	startHandle, endHandle := reorgInfo.StartHandle, reorgInfo.EndHandle
+	sessCtx := newContext(reorgInfo.d.store)
+	decodeColMap, err := makeupDecodeColMap(sessCtx, t, indexInfo)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	// variable.ddlReorgWorkerCounter can be modified by system variable "tidb_ddl_reorg_worker_cnt".
+	workerCnt := variable.GetDDLReorgWorkerCounter()
+	idxWorkers := make([]*addIndexWorker, 0, workerCnt)
+	defer func() {
+		closeAddIndexWorkers(idxWorkers)
+	}()
+
 	for {
 		kvRanges, err := splitTableRanges(t, reorgInfo.d.store, startHandle, endHandle)
 		if err != nil {
 			return errors.Trace(err)
 		}
 
+		// For dynamic adjust add index worker number.
+		workerCnt = variable.GetDDLReorgWorkerCounter()
+		// If only have 1 range, we can only start 1 worker.
+		if len(kvRanges) < int(workerCnt) {
+			workerCnt = int32(len(kvRanges))
+		}
+		// Enlarge the worker size.
+		for i := len(idxWorkers); i < int(workerCnt); i++ {
+			sessCtx := newContext(reorgInfo.d.store)
+			idxWorker := newAddIndexWorker(sessCtx, w, i, t, indexInfo, decodeColMap)
+			idxWorker.priority = job.Priority
+			idxWorkers = append(idxWorkers, idxWorker)
+			go idxWorkers[i].run(reorgInfo.d)
+		}
+		// Shrink the worker size.
+		if len(idxWorkers) > int(workerCnt) {
+			workers := idxWorkers[workerCnt:]
+			idxWorkers = idxWorkers[:workerCnt]
+			closeAddIndexWorkers(workers)
+		}
+
+		reorgInfo.d.mu.Lock()
+		reorgInfo.d.mu.hook.OnIndexWorkerReorgBefore(len(idxWorkers), len(kvRanges))
+		reorgInfo.d.mu.Unlock()
+
 		log.Infof("[ddl-reorg] start to reorg index of %v region ranges, handle range:[%v, %v).", len(kvRanges), startHandle, endHandle)
-		remains, err := w.sendRangeTaskToWorkers(t, workers, reorgInfo, &totalAddedCount, kvRanges)
+		remains, err := w.sendRangeTaskToWorkers(t, idxWorkers, reorgInfo, &totalAddedCount, kvRanges)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -1081,23 +1119,7 @@ func (w *worker) buildIndexForReorgInfo(t table.PhysicalTable, workers []*addInd
 func (w *worker) addPhysicalTableIndex(t table.PhysicalTable, indexInfo *model.IndexInfo, reorgInfo *reorgInfo) error {
 	job := reorgInfo.Job
 	log.Infof("[ddl-reorg] addTableIndex, job:%s, reorgInfo:%#v", job, reorgInfo)
-	sessCtx := newContext(reorgInfo.d.store)
-	decodeColMap, err := makeupDecodeColMap(sessCtx, t, indexInfo)
-	if err != nil {
-		return errors.Trace(err)
-	}
-
-	// variable.ddlReorgWorkerCounter can be modified by system variable "tidb_ddl_reorg_worker_cnt".
-	workerCnt := variable.GetDDLReorgWorkerCounter()
-	idxWorkers := make([]*addIndexWorker, workerCnt)
-	for i := 0; i < int(workerCnt); i++ {
-		sessCtx := newContext(reorgInfo.d.store)
-		idxWorkers[i] = newAddIndexWorker(sessCtx, w, i, t, indexInfo, decodeColMap)
-		idxWorkers[i].priority = job.Priority
-		go idxWorkers[i].run(reorgInfo.d)
-	}
-	defer closeAddIndexWorkers(idxWorkers)
-	err = w.buildIndexForReorgInfo(t, idxWorkers, job, reorgInfo)
+	err := w.buildIndexForReorgInfo(t, indexInfo, job, reorgInfo)
 	return errors.Trace(err)
 }