config, sysvar: add config instance.enable_ddl and sysvar tidb_enable_ddl

br/cmd/br/backup.go

@@ -11,7 +11,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/trace"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/br/pkg/version/build"
-	"github.com/pingcap/tidb/ddl"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/session"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
@@ -78,7 +78,7 @@ func NewBackupCommand() *cobra.Command {
 			task.LogArguments(c)
 
 			// Do not run ddl worker in BR.
-			ddl.RunWorker = false
+			config.GetGlobalConfig().Instance.TiDBEnableDDL.Store(false)
 
 			summary.SetUnit(summary.BackupUnit)
 			return nil

cmd/ddltest/ddl_test.go

@@ -33,7 +33,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	"github.com/pingcap/tidb/ddl"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/model"
@@ -116,7 +116,7 @@ func createDDLSuite(t *testing.T) (s *ddlSuite) {
 	// Stop current DDL worker, so that we can't be the owner now.
 	err = domain.GetDomain(s.ctx).DDL().Stop()
 	require.NoError(t, err)
-	ddl.RunWorker = false
+	config.GetGlobalConfig().Instance.TiDBEnableDDL.Store(false)
 	session.ResetStoreForWithTiKVTest(s.store)
 	s.dom.Close()
 	require.NoError(t, s.store.Close())

config/config.go

@@ -118,6 +118,7 @@ var (
 				"check-mb4-value-in-utf8":       "tidb_check_mb4_value_in_utf8",
 				"enable-collect-execution-info": "tidb_enable_collect_execution_info",
 				"max-server-connections":        "max_connections",
+				"run-ddl":                       "tidb_enable_ddl",
 			},
 		},
 		{
@@ -472,10 +473,11 @@ type Instance struct {
 	ForcePriority         string     `toml:"tidb_force_priority" json:"tidb_force_priority"`
 	MemoryUsageAlarmRatio float64    `toml:"tidb_memory_usage_alarm_ratio" json:"tidb_memory_usage_alarm_ratio"`
 	// EnableCollectExecutionInfo enables the TiDB to collect execution info.
-	EnableCollectExecutionInfo bool   `toml:"tidb_enable_collect_execution_info" json:"tidb_enable_collect_execution_info"`
-	PluginDir                  string `toml:"plugin_dir" json:"plugin_dir"`
-	PluginLoad                 string `toml:"plugin_load" json:"plugin_load"`
-	MaxConnections             uint32 `toml:"max_connections" json:"max_connections"`
+	EnableCollectExecutionInfo bool       `toml:"tidb_enable_collect_execution_info" json:"tidb_enable_collect_execution_info"`
+	PluginDir                  string     `toml:"plugin_dir" json:"plugin_dir"`
+	PluginLoad                 string     `toml:"plugin_load" json:"plugin_load"`
+	MaxConnections             uint32     `toml:"max_connections" json:"max_connections"`
+	TiDBEnableDDL              AtomicBool `toml:"tidb_enable_ddl" json:"tidb_enable_ddl"`
 }
 
 func (l *Log) getDisableTimestamp() bool {
@@ -853,6 +855,7 @@ var defaultConf = Config{
 		PluginDir:                   "/data/deploy/plugin",
 		PluginLoad:                  "",
 		MaxConnections:              0,
+		TiDBEnableDDL:               *NewAtomicBool(true),
 	},
 	Status: Status{
 		ReportStatus:          true,
@@ -1188,7 +1191,7 @@ func (c *Config) Valid() error {
 		}
 		return fmt.Errorf("invalid store=%s, valid storages=%v", c.Store, nameList)
 	}
-	if c.Store == "mocktikv" && !c.RunDDL {
+	if c.Store == "mocktikv" && !c.Instance.TiDBEnableDDL.Load() {
 		return fmt.Errorf("can't disable DDL on mocktikv")
 	}
 	if c.MaxIndexLength < DefMaxIndexLength || c.MaxIndexLength > DefMaxOfMaxIndexLength {

config/config.toml.example

@@ -18,9 +18,6 @@ path = "/tmp/tidb"
 # The socket file to use for connection.
 socket = "/tmp/tidb-{Port}.sock"
 
-# Run ddl worker on this tidb-server.
-run-ddl = true
-
 # Schema lease duration, very dangerous to change only if you know what you do.
 lease = "45s"
 
@@ -468,3 +465,6 @@ tidb_record_plan_in_slow_log = 1
 
 # The maximum permitted number of simultaneous client connections. When the value is 0, the number of connections is unlimited.
 max_connections = 0
+
+# Run ddl worker on this tidb-server.
+tidb_enable_ddl = true

config/config_test.go

@@ -196,9 +196,6 @@ path = "/tmp/tidb"
 # The socket file to use for connection.
 socket = "/tmp/tidb-{Port}.sock"
 
-# Run ddl worker on this tidb-server.
-run-ddl = true
-
 # Schema lease duration, very dangerous to change only if you know what you do.
 lease = "45s"
 
@@ -311,6 +308,9 @@ enable-enum-length-limit = true
 # The maximum permitted number of simultaneous client connections. When the value is 0, the number of connections is unlimited.
 max_connections = 0
 
+# Run ddl worker on this tidb-server.
+tidb_enable_ddl = true
+
 [log]
 # Log level: debug, info, warn, error, fatal.
 level = "info"
@@ -1023,7 +1023,10 @@ func TestConflictInstanceConfig(t *testing.T) {
 	// Just receive a warning and keep their respective values.
 	expectedConflictOptions := map[string]InstanceConfigSection{
 		"": {
-			"", map[string]string{"check-mb4-value-in-utf8": "tidb_check_mb4_value_in_utf8"},
+			"", map[string]string{
+				"check-mb4-value-in-utf8": "tidb_check_mb4_value_in_utf8",
+				"run-ddl":                 "tidb_enable_ddl",
+			},
 		},
 		"log": {
 			"log", map[string]string{"enable-slow-log": "tidb_enable_slow_log"},
@@ -1032,10 +1035,10 @@ func TestConflictInstanceConfig(t *testing.T) {
 			"performance", map[string]string{"force-priority": "tidb_force_priority"},
 		},
 	}
-	_, err = f.WriteString("check-mb4-value-in-utf8 = true \n" +
+	_, err = f.WriteString("check-mb4-value-in-utf8 = true \nrun-ddl = true \n" +
 		"[log] \nenable-slow-log = true \n" +
 		"[performance] \nforce-priority = \"NO_PRIORITY\"\n" +
-		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"")
+		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"\ntidb_enable_ddl = false")
 	require.NoError(t, err)
 	require.NoError(t, f.Sync())
 	err = conf.Load(configFile)
@@ -1047,6 +1050,8 @@ func TestConflictInstanceConfig(t *testing.T) {
 	require.Equal(t, false, conf.Instance.EnableSlowLog.Load())
 	require.Equal(t, "NO_PRIORITY", conf.Performance.ForcePriority)
 	require.Equal(t, "LOW_PRIORITY", conf.Instance.ForcePriority)
+	require.Equal(t, true, conf.RunDDL)
+	require.Equal(t, false, conf.Instance.TiDBEnableDDL.Load())
 	require.Equal(t, 0, len(DeprecatedOptions))
 	for _, conflictOption := range ConflictOptions {
 		expectedConflictOption, ok := expectedConflictOptions[conflictOption.SectionName]
@@ -1076,6 +1081,7 @@ func TestDeprecatedConfig(t *testing.T) {
 		"": {
 			"", map[string]string{
 				"enable-collect-execution-info": "tidb_enable_collect_execution_info",
+				"run-ddl":                       "tidb_enable_ddl",
 			},
 		},
 		"log": {
@@ -1091,7 +1097,7 @@ func TestDeprecatedConfig(t *testing.T) {
 			},
 		},
 	}
-	_, err = f.WriteString("enable-collect-execution-info = false \n" +
+	_, err = f.WriteString("enable-collect-execution-info = false \nrun-ddl = false \n" +
 		"[plugin] \ndir=\"/plugin-path\" \nload=\"audit-1,whitelist-1\" \n" +
 		"[log] \nslow-threshold = 100 \n" +
 		"[performance] \nmemory-usage-alarm-ratio = 0.5")

ddl/ddl.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl/util"
+	"github.com/pingcap/tidb/domain/infosync"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
@@ -439,6 +440,9 @@ func newDDL(ctx context.Context, options ...Option) *ddl {
 		limitJobCh:        make(chan *limitJobTask, batchAddingJobs),
 		enableTiFlashPoll: atomicutil.NewBool(true),
 	}
+	// Register functions for enable/disable ddl when changing system variable `tidb_enable_ddl`.
+	variable.EnableDDL = d.EnableDDL
+	variable.DisableDDL = d.DisableDDL
 
 	return d
 }
@@ -468,54 +472,91 @@ func (d *ddl) newDeleteRangeManager(mock bool) delRangeManager {
 
 // Start implements DDL.Start interface.
 func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
-	logutil.BgLogger().Info("[ddl] start DDL", zap.String("ID", d.uuid), zap.Bool("runWorker", RunWorker))
+	var err error
+	logutil.BgLogger().Info("[ddl] start DDL", zap.String("ID", d.uuid), zap.Bool("runWorker", config.GetGlobalConfig().Instance.TiDBEnableDDL.Load()))
 
 	d.wg.Run(d.limitDDLJobs)
 	d.sessPool = newSessionPool(ctxPool, d.store)
+	d.workers = make(map[workerType]*worker, 2)
+	d.delRangeMgr = d.newDeleteRangeManager(ctxPool == nil)
+	d.workers[generalWorker] = newWorker(d.ctx, generalWorker, d.sessPool, d.delRangeMgr, d.ddlCtx)
+	d.workers[addIdxWorker] = newWorker(d.ctx, addIdxWorker, d.sessPool, d.delRangeMgr, d.ddlCtx)
+	for _, worker := range d.workers {
+		worker.wg.Add(1)
+		w := worker
+		go w.start(d.ddlCtx)
+
+		metrics.DDLCounter.WithLabelValues(fmt.Sprintf("%s_%s", metrics.CreateDDL, worker.String())).Inc()
+
+		// When the start function is called, we will send a fake job to let worker
+		// checks owner firstly and try to find whether a job exists and run.
+		asyncNotify(worker.ddlJobCh)
+	}
+
+	d.ddlSeqNumMu.seqNum, err = d.GetNextDDLSeqNum()
+	if err != nil {
+		return err
+	}
+
+	go d.schemaSyncer.StartCleanWork()
+	if config.TableLockEnabled() {
+		d.wg.Add(1)
+		go d.startCleanDeadTableLock()
+	}
+	metrics.DDLCounter.WithLabelValues(metrics.StartCleanWork).Inc()
 
-	// If RunWorker is true, we need campaign owner and do DDL job.
+	// If tidb_enable_ddl is true, we need campaign owner and do DDL job.
 	// Otherwise, we needn't do that.
-	if RunWorker {
-		err := d.ownerManager.CampaignOwner()
+	if config.GetGlobalConfig().Instance.TiDBEnableDDL.Load() {
+		err = d.ownerManager.CampaignOwner()
 		if err != nil {
 			return errors.Trace(err)
 		}
+	}
 
-		d.workers = make(map[workerType]*worker, 2)
-		d.delRangeMgr = d.newDeleteRangeManager(ctxPool == nil)
-		d.workers[generalWorker] = newWorker(d.ctx, generalWorker, d.sessPool, d.delRangeMgr, d.ddlCtx)
-		d.workers[addIdxWorker] = newWorker(d.ctx, addIdxWorker, d.sessPool, d.delRangeMgr, d.ddlCtx)
-		for _, worker := range d.workers {
-			worker.wg.Add(1)
-			w := worker
-			go w.start(d.ddlCtx)
+	variable.RegisterStatistics(d)
 
-			metrics.DDLCounter.WithLabelValues(fmt.Sprintf("%s_%s", metrics.CreateDDL, worker.String())).Inc()
+	metrics.DDLCounter.WithLabelValues(metrics.CreateDDLInstance).Inc()
 
-			// When the start function is called, we will send a fake job to let worker
-			// checks owner firstly and try to find whether a job exists and run.
-			asyncNotify(worker.ddlJobCh)
-		}
+	// Start some background routine to manage TiFlash replica.
+	d.wg.Run(d.PollTiFlashRoutine)
 
-		d.ddlSeqNumMu.seqNum, err = d.GetNextDDLSeqNum()
+	return nil
+}
+
+// EnableDDL enable this node to execute ddl.
+// Since ownerManager.CampaignOwner will start a new goroutine to run ownerManager.campaignLoop,
+// we should make sure that before invoking EnableDDL(), ddl is DISABLE.
+func (d *ddl) EnableDDL() error {
+	err := d.ownerManager.CampaignOwner()
+	return errors.Trace(err)
+}
+
+// DisableDDL disable this node to execute ddl.
+// We should make sure that before invoking DisableDDL(), ddl is ENABLE.
+func (d *ddl) DisableDDL() error {
+	if d.ownerManager.IsOwner() {
+		// If there is only one node, we should NOT disable ddl.
+		serverInfo, err := infosync.GetAllServerInfo(d.ctx)
 		if err != nil {
 			return err
 		}
-
-		go d.schemaSyncer.StartCleanWork()
-		if config.TableLockEnabled() {
-			d.wg.Add(1)
-			go d.startCleanDeadTableLock()
+		if len(serverInfo) <= 1 {
+			return dbterror.ErrDDLSetting
 		}
-		metrics.DDLCounter.WithLabelValues(metrics.StartCleanWork).Inc()
-	}
 
-	variable.RegisterStatistics(d)
+		// FIXME: if possible, when this node is the only node with DDL, ths setting of DisableDDL should fail.
 
-	metrics.DDLCounter.WithLabelValues(metrics.CreateDDLInstance).Inc()
+		// lets the owner start a new election.
+		err = d.ownerManager.ResignOwner(d.ctx)
 
-	// Start some background routine to manage TiFlash replica.
-	d.wg.Run(d.PollTiFlashRoutine)
+		if err != nil {
+			return err
+		}
+	}
+
+	// disable campaign by interrupting campaignLoop
+	d.ownerManager.CampaignCancel()
 
 	return nil
 }
@@ -661,7 +702,7 @@ func getJobCheckInterval(job *model.Job, i int) (time.Duration, bool) {
 
 func (d *ddl) asyncNotifyWorker(job *model.Job) {
 	// If the workers don't run, we needn't notify workers.
-	if !RunWorker {
+	if !config.GetGlobalConfig().Instance.TiDBEnableDDL.Load() {
 		return
 	}
 	var worker *worker

ddl/ddl_worker.go

@@ -49,8 +49,6 @@ import (
 )
 
 var (
-	// RunWorker indicates if this TiDB server starts DDL worker and can run DDL job.
-	RunWorker = true
 	// ddlWorkerID is used for generating the next DDL worker ID.
 	ddlWorkerID = atomicutil.NewInt32(0)
 	// WaitTimeWhenErrorOccurred is waiting interval when processing DDL jobs encounter errors.

ddl/restart_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/parser/model"
@@ -48,7 +49,7 @@ func (d *ddl) restartWorkers(ctx context.Context) {
 	d.ctx, d.cancel = context.WithCancel(ctx)
 
 	d.wg.Run(d.limitDDLJobs)
-	if !RunWorker {
+	if !config.GetGlobalConfig().Instance.TiDBEnableDDL.Load() {
 		return
 	}
 

errno/errcode.go

@@ -1072,6 +1072,8 @@ const (
 	ErrHTTPServiceError                   = 8243
 	ErrPartitionColumnStatsMissing        = 8244
 	ErrColumnInChange                     = 8245
+	ErrDDLSetting                         = 8246
+
 	// TiKV/PD/TiFlash errors.
 	ErrPDServerTimeout           = 9001
 	ErrTiKVServerTimeout         = 9002

errno/errname.go

@@ -1065,6 +1065,7 @@ var MySQLErrName = map[uint16]*mysql.ErrMessage{
 	ErrJSONObjectKeyTooLong:        mysql.Message("TiDB does not yet support JSON objects with the key length >= 65536", nil),
 	ErrPartitionStatsMissing:       mysql.Message("Build table: %s global-level stats failed due to missing partition-level stats", nil),
 	ErrPartitionColumnStatsMissing: mysql.Message("Build table: %s global-level stats failed due to missing partition-level column stats, please run analyze table to refresh columns of all partitions", nil),
+	ErrDDLSetting:                  mysql.Message("Error happened when enable/disable DDL", nil),
 	ErrNotSupportedWithSem:         mysql.Message("Feature '%s' is not supported when security enhanced mode is enabled", nil),
 
 	ErrPlacementPolicyCheck:            mysql.Message("Placement policy didn't meet the constraint, reason: %s", nil),

errors.toml

@@ -1236,6 +1236,11 @@ error = '''
 column %s id %d does not exist, this column may have been updated by other DDL ran in parallel
 '''
 
+["ddl:8246"]
+error = '''
+Error happened when enable/disable DDL
+'''
+
 ["domain:8027"]
 error = '''
 Information schema is out of date: schema failed to update in 1 lease, please make sure TiDB can connect to TiKV