Skip to content

Commit 0280bcd

Browse files
crazycs520zz-jason
crazycs520
authored and
zz-jason
committed
variable: remove tidb_ddl_reorg_worker_cnt and tidb_ddl_reorg_batch_size session level. (#9134)
1 parent d842dc3 commit 0280bcd

File tree

8 files changed

+91
-31
lines changed

8 files changed

+91
-31
lines changed

‎ddl/fail_db_test.go

+5-2
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"github.com/pingcap/parser"
2626
"github.com/pingcap/parser/model"
2727
"github.com/pingcap/tidb/ddl"
28+
ddlutil "github.com/pingcap/tidb/ddl/util"
2829
"github.com/pingcap/tidb/domain"
2930
"github.com/pingcap/tidb/kv"
3031
"github.com/pingcap/tidb/session"
@@ -269,11 +270,13 @@ func (s *testDBSuite) TestAddIndexWorkerNum(c *C) {
269270
// Split table to multi region.
270271
s.cluster.SplitTable(s.mvccStore, tbl.Meta().ID, splitCount)
271272

273+
err = ddlutil.LoadDDLReorgVars(tk.Se)
274+
c.Assert(err, IsNil)
272275
originDDLAddIndexWorkerCnt := variable.GetDDLReorgWorkerCounter()
273276
lastSetWorkerCnt := originDDLAddIndexWorkerCnt
274277
atomic.StoreInt32(&ddl.TestCheckWorkerNumber, lastSetWorkerCnt)
275278
ddl.TestCheckWorkerNumber = lastSetWorkerCnt
276-
defer variable.SetDDLReorgWorkerCounter(originDDLAddIndexWorkerCnt)
279+
defer tk.MustExec(fmt.Sprintf("set @@global.tidb_ddl_reorg_worker_cnt=%d", originDDLAddIndexWorkerCnt))
277280

278281
gofail.Enable("github.com/pingcap/tidb/ddl/checkIndexWorkerNum", `return(true)`)
279282
defer gofail.Disable("github.com/pingcap/tidb/ddl/checkIndexWorkerNum")
@@ -291,7 +294,7 @@ LOOP:
291294
c.Assert(err, IsNil, Commentf("err:%v", errors.ErrorStack(err)))
292295
case <-ddl.TestCheckWorkerNumCh:
293296
lastSetWorkerCnt = int32(rand.Intn(8) + 8)
294-
tk.MustExec(fmt.Sprintf("set @@tidb_ddl_reorg_worker_cnt=%d", lastSetWorkerCnt))
297+
tk.MustExec(fmt.Sprintf("set @@global.tidb_ddl_reorg_worker_cnt=%d", lastSetWorkerCnt))
295298
atomic.StoreInt32(&ddl.TestCheckWorkerNumber, lastSetWorkerCnt)
296299
checkNum++
297300
}

‎ddl/index.go

+15
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"github.com/pingcap/parser/ast"
2424
"github.com/pingcap/parser/model"
2525
"github.com/pingcap/parser/mysql"
26+
ddlutil "github.com/pingcap/tidb/ddl/util"
2627
"github.com/pingcap/tidb/expression"
2728
"github.com/pingcap/tidb/infoschema"
2829
"github.com/pingcap/tidb/kv"
@@ -1052,6 +1053,17 @@ var (
10521053
TestCheckWorkerNumber = int32(16)
10531054
)
10541055

1056+
func loadDDLReorgVars(w *worker) error {
1057+
// Get sessionctx from context resource pool.
1058+
var ctx sessionctx.Context
1059+
ctx, err := w.sessPool.get()
1060+
if err != nil {
1061+
return errors.Trace(err)
1062+
}
1063+
defer w.sessPool.put(ctx)
1064+
return ddlutil.LoadDDLReorgVars(ctx)
1065+
}
1066+
10551067
// addPhysicalTableIndex handles the add index reorganization state for a non-partitioned table or a partition.
10561068
// For a partitioned table, it should be handled partition by partition.
10571069
//
@@ -1092,6 +1104,9 @@ func (w *worker) addPhysicalTableIndex(t table.PhysicalTable, indexInfo *model.I
10921104
}
10931105

10941106
// For dynamic adjust add index worker number.
1107+
if err := loadDDLReorgVars(w); err != nil {
1108+
log.Error(err)
1109+
}
10951110
workerCnt = variable.GetDDLReorgWorkerCounter()
10961111
// If only have 1 range, we can only start 1 worker.
10971112
if len(kvRanges) < int(workerCnt) {

‎ddl/util/util.go

+21
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"github.com/pingcap/parser/terror"
2222
"github.com/pingcap/tidb/kv"
2323
"github.com/pingcap/tidb/sessionctx"
24+
"github.com/pingcap/tidb/sessionctx/variable"
2425
"github.com/pingcap/tidb/util/chunk"
2526
"github.com/pingcap/tidb/util/sqlexec"
2627
"golang.org/x/net/context"
@@ -128,3 +129,23 @@ func UpdateDeleteRange(ctx sessionctx.Context, dr DelRangeTask, newStartKey, old
128129
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
129130
return errors.Trace(err)
130131
}
132+
133+
const loadDDLReorgVarsSQL = "select HIGH_PRIORITY variable_name, variable_value from mysql.global_variables where variable_name in ('" +
134+
variable.TiDBDDLReorgWorkerCount + "', '" +
135+
variable.TiDBDDLReorgBatchSize + "')"
136+
137+
// LoadDDLReorgVars loads ddl reorg variable from mysql.global_variables.
138+
func LoadDDLReorgVars(ctx sessionctx.Context) error {
139+
if sctx, ok := ctx.(sqlexec.RestrictedSQLExecutor); ok {
140+
rows, _, err := sctx.ExecRestrictedSQL(ctx, loadDDLReorgVarsSQL)
141+
if err != nil {
142+
return errors.Trace(err)
143+
}
144+
for _, row := range rows {
145+
varName := row.GetString(0)
146+
varValue := row.GetString(1)
147+
variable.SetLocalSystemVar(varName, varValue)
148+
}
149+
}
150+
return nil
151+
}

‎executor/ddl_test.go

+36-16
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"github.com/pingcap/parser/model"
2424
"github.com/pingcap/parser/mysql"
2525
"github.com/pingcap/parser/terror"
26+
ddlutil "github.com/pingcap/tidb/ddl/util"
2627
"github.com/pingcap/tidb/domain"
2728
"github.com/pingcap/tidb/meta/autoid"
2829
plannercore "github.com/pingcap/tidb/planner/core"
@@ -447,24 +448,32 @@ func (s *testSuite) TestMaxHandleAddIndex(c *C) {
447448
func (s *testSuite) TestSetDDLReorgWorkerCnt(c *C) {
448449
tk := testkit.NewTestKit(c, s.store)
449450
tk.MustExec("use test")
451+
err := ddlutil.LoadDDLReorgVars(tk.Se)
452+
c.Assert(err, IsNil)
450453
c.Assert(variable.GetDDLReorgWorkerCounter(), Equals, int32(variable.DefTiDBDDLReorgWorkerCount))
451-
tk.MustExec("set tidb_ddl_reorg_worker_cnt = 1")
454+
tk.MustExec("set @@global.tidb_ddl_reorg_worker_cnt = 1")
455+
err = ddlutil.LoadDDLReorgVars(tk.Se)
456+
c.Assert(err, IsNil)
452457
c.Assert(variable.GetDDLReorgWorkerCounter(), Equals, int32(1))
453-
tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
458+
tk.MustExec("set @@global.tidb_ddl_reorg_worker_cnt = 100")
459+
err = ddlutil.LoadDDLReorgVars(tk.Se)
460+
c.Assert(err, IsNil)
454461
c.Assert(variable.GetDDLReorgWorkerCounter(), Equals, int32(100))
455-
_, err := tk.Exec("set tidb_ddl_reorg_worker_cnt = invalid_val")
462+
_, err = tk.Exec("set @@global.tidb_ddl_reorg_worker_cnt = invalid_val")
456463
c.Assert(terror.ErrorEqual(err, variable.ErrWrongTypeForVar), IsTrue, Commentf("err %v", err))
457-
tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
464+
tk.MustExec("set @@global.tidb_ddl_reorg_worker_cnt = 100")
465+
err = ddlutil.LoadDDLReorgVars(tk.Se)
466+
c.Assert(err, IsNil)
458467
c.Assert(variable.GetDDLReorgWorkerCounter(), Equals, int32(100))
459-
_, err = tk.Exec("set tidb_ddl_reorg_worker_cnt = -1")
468+
_, err = tk.Exec("set @@global.tidb_ddl_reorg_worker_cnt = -1")
460469
c.Assert(terror.ErrorEqual(err, variable.ErrWrongValueForVar), IsTrue, Commentf("err %v", err))
461470

462-
tk.MustExec("set tidb_ddl_reorg_worker_cnt = 100")
463-
res := tk.MustQuery("select @@tidb_ddl_reorg_worker_cnt")
471+
tk.MustExec("set @@global.tidb_ddl_reorg_worker_cnt = 100")
472+
res := tk.MustQuery("select @@global.tidb_ddl_reorg_worker_cnt")
464473
res.Check(testkit.Rows("100"))
465474

466475
res = tk.MustQuery("select @@global.tidb_ddl_reorg_worker_cnt")
467-
res.Check(testkit.Rows(fmt.Sprintf("%v", variable.DefTiDBDDLReorgWorkerCount)))
476+
res.Check(testkit.Rows("100"))
468477
tk.MustExec("set @@global.tidb_ddl_reorg_worker_cnt = 100")
469478
res = tk.MustQuery("select @@global.tidb_ddl_reorg_worker_cnt")
470479
res.Check(testkit.Rows("100"))
@@ -473,28 +482,39 @@ func (s *testSuite) TestSetDDLReorgWorkerCnt(c *C) {
473482
func (s *testSuite) TestSetDDLReorgBatchSize(c *C) {
474483
tk := testkit.NewTestKit(c, s.store)
475484
tk.MustExec("use test")
485+
err := ddlutil.LoadDDLReorgVars(tk.Se)
486+
c.Assert(err, IsNil)
476487
c.Assert(variable.GetDDLReorgBatchSize(), Equals, int32(variable.DefTiDBDDLReorgBatchSize))
477488

478-
tk.MustExec("set tidb_ddl_reorg_batch_size = 1")
489+
tk.MustExec("set @@global.tidb_ddl_reorg_batch_size = 1")
479490
tk.MustQuery("show warnings;").Check(testkit.Rows("Warning 1292 Truncated incorrect tidb_ddl_reorg_batch_size value: '1'"))
491+
err = ddlutil.LoadDDLReorgVars(tk.Se)
492+
c.Assert(err, IsNil)
480493
c.Assert(variable.GetDDLReorgBatchSize(), Equals, int32(variable.MinDDLReorgBatchSize))
481-
tk.MustExec(fmt.Sprintf("set tidb_ddl_reorg_batch_size = %v", variable.MaxDDLReorgBatchSize+1))
494+
tk.MustExec(fmt.Sprintf("set @@global.tidb_ddl_reorg_batch_size = %v", variable.MaxDDLReorgBatchSize+1))
482495
tk.MustQuery("show warnings;").Check(testkit.Rows(fmt.Sprintf("Warning 1292 Truncated incorrect tidb_ddl_reorg_batch_size value: '%d'", variable.MaxDDLReorgBatchSize+1)))
496+
err = ddlutil.LoadDDLReorgVars(tk.Se)
497+
c.Assert(err, IsNil)
483498
c.Assert(variable.GetDDLReorgBatchSize(), Equals, int32(variable.MaxDDLReorgBatchSize))
484-
_, err := tk.Exec("set tidb_ddl_reorg_batch_size = invalid_val")
499+
_, err = tk.Exec("set @@global.tidb_ddl_reorg_batch_size = invalid_val")
485500
c.Assert(terror.ErrorEqual(err, variable.ErrWrongTypeForVar), IsTrue, Commentf("err %v", err))
486-
tk.MustExec("set tidb_ddl_reorg_batch_size = 100")
501+
tk.MustExec("set @@global.tidb_ddl_reorg_batch_size = 100")
502+
err = ddlutil.LoadDDLReorgVars(tk.Se)
503+
c.Assert(err, IsNil)
487504
c.Assert(variable.GetDDLReorgBatchSize(), Equals, int32(100))
488-
tk.MustExec("set tidb_ddl_reorg_batch_size = -1")
505+
tk.MustExec("set @@global.tidb_ddl_reorg_batch_size = -1")
489506
tk.MustQuery("show warnings;").Check(testkit.Rows("Warning 1292 Truncated incorrect tidb_ddl_reorg_batch_size value: '-1'"))
490507

491-
tk.MustExec("set tidb_ddl_reorg_batch_size = 100")
492-
res := tk.MustQuery("select @@tidb_ddl_reorg_batch_size")
508+
tk.MustExec("set @@global.tidb_ddl_reorg_batch_size = 100")
509+
res := tk.MustQuery("select @@global.tidb_ddl_reorg_batch_size")
493510
res.Check(testkit.Rows("100"))
494511

495512
res = tk.MustQuery("select @@global.tidb_ddl_reorg_batch_size")
496-
res.Check(testkit.Rows(fmt.Sprintf("%v", variable.DefTiDBDDLReorgBatchSize)))
513+
res.Check(testkit.Rows(fmt.Sprintf("%v", 100)))
497514
tk.MustExec("set @@global.tidb_ddl_reorg_batch_size = 1000")
498515
res = tk.MustQuery("select @@global.tidb_ddl_reorg_batch_size")
499516
res.Check(testkit.Rows("1000"))
517+
518+
// If do not LoadDDLReorgVars, the local variable will be the last loaded value.
519+
c.Assert(variable.GetDDLReorgBatchSize(), Equals, int32(100))
500520
}

‎session/session.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -749,8 +749,9 @@ func (s *session) SetGlobalSysVar(name, value string) error {
749749
if err != nil {
750750
return errors.Trace(err)
751751
}
752+
name = strings.ToLower(name)
752753
sql := fmt.Sprintf(`REPLACE %s.%s VALUES ('%s', '%s');`,
753-
mysql.SystemDB, mysql.GlobalVariablesTable, strings.ToLower(name), sVal)
754+
mysql.SystemDB, mysql.GlobalVariablesTable, name, sVal)
754755
_, _, err = s.ExecRestrictedSQL(s, sql)
755756
return errors.Trace(err)
756757
}
@@ -1352,9 +1353,7 @@ const loadCommonGlobalVarsSQL = "select HIGH_PRIORITY * from mysql.global_variab
13521353
variable.TiDBHashAggPartialConcurrency + quoteCommaQuote +
13531354
variable.TiDBHashAggFinalConcurrency + quoteCommaQuote +
13541355
variable.TiDBBackoffLockFast + quoteCommaQuote +
1355-
variable.TiDBDDLReorgWorkerCount + quoteCommaQuote +
13561356
variable.TiDBOptInSubqUnFolding + quoteCommaQuote +
1357-
variable.TiDBDDLReorgBatchSize + quoteCommaQuote +
13581357
variable.TiDBDistSQLScanConcurrency + quoteCommaQuote +
13591358
variable.TiDBMaxChunkSize + quoteCommaQuote +
13601359
variable.TiDBRetryLimit + quoteCommaQuote +

‎sessionctx/variable/session.go

+10-4
Original file line numberDiff line numberDiff line change
@@ -597,10 +597,6 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
597597
s.OptimizerSelectivityLevel = tidbOptPositiveInt32(val, DefTiDBOptimizerSelectivityLevel)
598598
case TiDBEnableTablePartition:
599599
s.EnableTablePartition = TiDBOptOn(val)
600-
case TiDBDDLReorgWorkerCount:
601-
SetDDLReorgWorkerCounter(int32(tidbOptPositiveInt32(val, DefTiDBDDLReorgWorkerCount)))
602-
case TiDBDDLReorgBatchSize:
603-
SetDDLReorgBatchSize(int32(tidbOptPositiveInt32(val, DefTiDBDDLReorgBatchSize)))
604600
case TiDBDDLReorgPriority:
605601
s.setDDLReorgPriority(val)
606602
case TiDBForcePriority:
@@ -610,6 +606,16 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
610606
return nil
611607
}
612608

609+
// SetLocalSystemVar sets values of the local variables which in "server" scope.
610+
func SetLocalSystemVar(name string, val string) {
611+
switch name {
612+
case TiDBDDLReorgWorkerCount:
613+
SetDDLReorgWorkerCounter(int32(tidbOptPositiveInt32(val, DefTiDBDDLReorgWorkerCount)))
614+
case TiDBDDLReorgBatchSize:
615+
SetDDLReorgBatchSize(int32(tidbOptPositiveInt32(val, DefTiDBDDLReorgBatchSize)))
616+
}
617+
}
618+
613619
// special session variables.
614620
const (
615621
SQLModeVar = "sql_mode"

‎sessionctx/variable/sysvar.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -664,8 +664,8 @@ var defaultSysVars = []*SysVar{
664664
{ScopeSession, TiDBSlowLogThreshold, strconv.Itoa(logutil.DefaultSlowThreshold)},
665665
{ScopeSession, TiDBQueryLogMaxLen, strconv.Itoa(logutil.DefaultQueryLogMaxLen)},
666666
{ScopeSession, TiDBConfig, ""},
667-
{ScopeGlobal | ScopeSession, TiDBDDLReorgWorkerCount, strconv.Itoa(DefTiDBDDLReorgWorkerCount)},
668-
{ScopeGlobal | ScopeSession, TiDBDDLReorgBatchSize, strconv.Itoa(DefTiDBDDLReorgBatchSize)},
667+
{ScopeGlobal, TiDBDDLReorgWorkerCount, strconv.Itoa(DefTiDBDDLReorgWorkerCount)},
668+
{ScopeGlobal, TiDBDDLReorgBatchSize, strconv.Itoa(DefTiDBDDLReorgBatchSize)},
669669
{ScopeSession, TiDBDDLReorgPriority, "PRIORITY_LOW"},
670670
{ScopeSession, TiDBForcePriority, mysql.Priority2Str[DefTiDBForcePriority]},
671671
}

‎sessionctx/variable/varsutil_test.go

-4
Original file line numberDiff line numberDiff line change
@@ -200,10 +200,6 @@ func (s *testVarsutilSuite) TestVarsutil(c *C) {
200200
SetSessionSystemVar(v, TiDBOptimizerSelectivityLevel, types.NewIntDatum(1))
201201
c.Assert(v.OptimizerSelectivityLevel, Equals, 1)
202202

203-
c.Assert(GetDDLReorgWorkerCounter(), Equals, int32(DefTiDBDDLReorgWorkerCount))
204-
SetSessionSystemVar(v, TiDBDDLReorgWorkerCount, types.NewIntDatum(1))
205-
c.Assert(GetDDLReorgWorkerCounter(), Equals, int32(1))
206-
207203
err = SetSessionSystemVar(v, TiDBDDLReorgWorkerCount, types.NewIntDatum(-1))
208204
c.Assert(terror.ErrorEqual(err, ErrWrongValueForVar), IsTrue)
209205

0 commit comments

Comments
 (0)