*: add session var 'tidb_ddl_reorg_worker_cnt' to control ddl reorg workers count

ddl/index.go

@@ -806,8 +806,8 @@ func (d *ddl) addTableIndex(t table.Table, indexInfo *model.IndexInfo, reorgInfo
 	log.Infof("[ddl-reorg] addTableIndex, job:%s, reorgInfo:%#v", job, reorgInfo)
 	colFieldMap := makeupIndexColFieldMap(t, indexInfo)
 
-	// TODO: make workerCnt can be modified by system variable.
-	workerCnt := defaultWorkers
+	// d.reorgWorkerCnt can be modified by system variable "tidb_ddl_reorg_worker_cnt".
+	workerCnt := d.workerVars.DDLReorgWorkerCount
 	workers := make([]*addIndexWorker, workerCnt)
 	for i := 0; i < workerCnt; i++ {
 		sessCtx := d.newContext()

executor/ddl_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/tidb/model"
 	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/plan"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
@@ -373,3 +374,25 @@ func (s *testSuite) TestShardRowIDBits(c *C) {
 	c.Assert(count, Equals, 100)
 	c.Assert(hasShardedID, IsTrue)
 }
+
+func (s *testSuite) TestSetDDLReorgWorkerCnt(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, variable.DefTiDBDDLReorgWorkerCount)
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = 1")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, 1)
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, 100)
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = invalid_val")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, variable.DefTiDBDDLReorgWorkerCount)
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, 100)
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = -1")
+	c.Assert(domain.GetDomain(tk.Se).DDL().WorkerVars().DDLReorgWorkerCount, Equals, variable.DefTiDBDDLReorgWorkerCount)
+
+	res := tk.MustQuery("select @@tidb_ddl_reorg_worker_cnt")
+	res.Check(testkit.Rows("-1"))
+	tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
+	res = tk.MustQuery("select @@tidb_ddl_reorg_worker_cnt")
+	res.Check(testkit.Rows("100"))
+}

executor/set.go

@@ -186,6 +186,8 @@ func (e *SetExecutor) setSysVariable(name string, v *expression.VarAssignment) e
 		if isoLevel == ast.ReadCommitted {
 			e.ctx.Txn().SetOption(kv.IsolationLevel, kv.RC)
 		}
+	} else if name == variable.TiDBDDLReorgWorkerCount {
+		domain.GetDomain(e.ctx).DDL().WorkerVars().DDLReorgWorkerCount = sessionVars.DDLReorgWorkerCount
 	}
 
 	return nil

sessionctx/variable/session.go

@@ -309,6 +309,7 @@ func NewSessionVars() *SessionVars {
 		IndexLookupJoinConcurrency: DefIndexLookupJoinConcurrency,
 		HashJoinConcurrency:        DefTiDBHashJoinConcurrency,
 		DistSQLScanConcurrency:     DefDistSQLScanConcurrency,
+		DDLReorgWorkerCount:        DefTiDBDDLReorgWorkerCount,
 	}
 	vars.MemQuota = MemQuota{
 		MemQuotaQuery:             DefTiDBMemQuotaQuery,
@@ -527,6 +528,8 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		s.EnableStreaming = TiDBOptOn(val)
 	case TiDBOptimizerSelectivityLevel:
 		s.OptimizerSelectivityLevel = tidbOptPositiveInt32(val, DefTiDBOptimizerSelectivityLevel)
+	case TiDBDDLReorgWorkerCount:
+		s.DDLReorgWorkerCount = tidbOptPositiveInt32(val, DefTiDBDDLReorgWorkerCount)
 	}
 	s.systems[name] = val
 	return nil
@@ -569,6 +572,9 @@ type Concurrency struct {
 
 	// IndexSerialScanConcurrency is the number of concurrent index serial scan worker.
 	IndexSerialScanConcurrency int
+
+	// DDLReorgWorkerCount defines the count of ddl reorganization workers.
+	DDLReorgWorkerCount int
 }
 
 // MemQuota defines memory quota values.

sessionctx/variable/sysvar.go

@@ -640,6 +640,7 @@ var defaultSysVars = []*SysVar{
 	/* The following variable is defined as session scope but is actually server scope. */
 	{ScopeSession, TiDBGeneralLog, strconv.Itoa(DefTiDBGeneralLog)},
 	{ScopeSession, TiDBConfig, ""},
+	{ScopeSession, TiDBDDLReorgWorkerCount, strconv.Itoa(DefTiDBDDLReorgWorkerCount)},
 }
 
 // SynonymsSysVariables is synonyms of system variables.

sessionctx/variable/tidb_vars.go

@@ -150,6 +150,9 @@ const (
 
 	// tidb_backoff_lock_fast is used for tikv backoff base time in milliseconds.
 	TiDBBackoffLockFast = "tidb_backoff_lock_fast"
+
+	// tidb_ddl_reorg_worker_cnt defines the count of ddl reorg workers.
+	TiDBDDLReorgWorkerCount = "tidb_ddl_reorg_worker_cnt"
 )
 
 // Default TiDB system variable values.
@@ -181,6 +184,7 @@ const (
 	DefTiDBGeneralLog                = 0
 	DefTiDBHashJoinConcurrency       = 5
 	DefTiDBOptimizerSelectivityLevel = 0
+	DefTiDBDDLReorgWorkerCount       = 16
 )
 
 // Process global variables.

sessionctx/variable/varsutil_test.go

@@ -216,6 +216,13 @@ func (s *testVarsutilSuite) TestVarsutil(c *C) {
 	c.Assert(v.OptimizerSelectivityLevel, Equals, DefTiDBOptimizerSelectivityLevel)
 	SetSessionSystemVar(v, TiDBOptimizerSelectivityLevel, types.NewIntDatum(1))
 	c.Assert(v.OptimizerSelectivityLevel, Equals, 1)
+
+	c.Assert(v.DDLReorgWorkerCount, Equals, DefTiDBDDLReorgWorkerCount)
+	SetSessionSystemVar(v, TiDBDDLReorgWorkerCount, types.NewIntDatum(1))
+	c.Assert(v.DDLReorgWorkerCount, Equals, 1)
+
+	SetSessionSystemVar(v, TiDBDDLReorgWorkerCount, types.NewIntDatum(-1))
+	c.Assert(v.DDLReorgWorkerCount, Equals, DefTiDBDDLReorgWorkerCount)
 }
 
 type mockGlobalAccessor struct {