*: support concurrent auto analyze (#54432)

close #53460

pkg/sessionctx/variable/sysvar.go

@@ -1398,6 +1398,18 @@ var defaultSysVars = []*SysVar{
 			return err
 		},
 	},
+	{Scope: ScopeGlobal, Name: TiDBAutoAnalyzeConcurrency, Value: strconv.Itoa(DefTiDBAutoAnalyzeConcurrency), Type: TypeInt, MinValue: 0, MaxValue: math.MaxInt32,
+		GetGlobal: func(_ context.Context, s *SessionVars) (string, error) {
+			return string(AutoAnlayzeConcurrency.Load()), nil
+		},
+		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
+			num, err := strconv.ParseInt(val, 10, 64)
+			if err == nil {
+				AutoAnlayzeConcurrency.Store(int32(num))
+			}
+			return err
+		},
+	},
 	{Scope: ScopeGlobal, Name: TiDBEnableMDL, Value: BoolToOnOff(DefTiDBEnableMDL), Type: TypeBool, SetGlobal: func(_ context.Context, vars *SessionVars, val string) error {
 		if EnableMDL.Load() != TiDBOptOn(val) {
 			err := SwitchMDL(TiDBOptOn(val))

pkg/sessionctx/variable/tidb_vars.go

@@ -1034,6 +1034,8 @@ const (
 	// TiDBMaxAutoAnalyzeTime is the max time that auto analyze can run. If auto analyze runs longer than the value, it
 	// will be killed. 0 indicates that there is no time limit.
 	TiDBMaxAutoAnalyzeTime = "tidb_max_auto_analyze_time"
+	// TiDBAutoAnalyzeConcurrency is the concurrency of the auto analyze
+	TiDBAutoAnalyzeConcurrency = "tidb_auto_analyze_concurrency"
 	// TiDBEnableDistTask indicates whether to enable the distributed execute background tasks(For example DDL, Import etc).
 	TiDBEnableDistTask = "tidb_enable_dist_task"
 	// TiDBEnableFastCreateTable indicates whether to enable the fast create table feature.
@@ -1407,6 +1409,7 @@ const (
 	DefTiDBAnalyzeColumnOptions                    = "PREDICATE"
 	DefTiDBMemOOMAction                            = "CANCEL"
 	DefTiDBMaxAutoAnalyzeTime                      = 12 * 60 * 60
+	DefTiDBAutoAnalyzeConcurrency                  = 2
 	DefTiDBEnablePrepPlanCache                     = true
 	DefTiDBPrepPlanCacheSize                       = 100
 	DefTiDBSessionPlanCacheSize                    = 100
@@ -1594,6 +1597,7 @@ var (
 	EnableNoopVariables               = atomic.NewBool(DefTiDBEnableNoopVariables)
 	EnableMDL                         = atomic.NewBool(false)
 	AutoAnalyzePartitionBatchSize     = atomic.NewInt64(DefTiDBAutoAnalyzePartitionBatchSize)
+	AutoAnlayzeConcurrency            = atomic.NewInt32(DefTiDBAutoAnalyzeConcurrency)
 	// EnableFastReorg indicates whether to use lightning to enhance DDL reorg performance.
 	EnableFastReorg = atomic.NewBool(DefTiDBEnableFastReorg)
 	// DDLDiskQuota is the temporary variable for set disk quota for lightning

pkg/statistics/handle/autoanalyze/refresher/refresher.go

@@ -86,6 +86,9 @@ func (r *Refresher) PickOneTableAndAnalyzeByPriority() bool {
 	}
 	defer r.statsHandle.SPool().Put(se)
 	sctx := se.(sessionctx.Context)
+	var wg util.WaitGroupWrapper
+	defer wg.Wait()
+	cnt := 0
 	// Pick the table with the highest weight.
 	for r.Jobs.Len() > 0 {
 		job := r.Jobs.Pop()
@@ -103,18 +106,25 @@ func (r *Refresher) PickOneTableAndAnalyzeByPriority() bool {
 			"Auto analyze triggered",
 			zap.Stringer("job", job),
 		)
-		err = job.Analyze(
-			r.statsHandle,
-			r.sysProcTracker,
-		)
-		if err != nil {
-			statslogutil.StatsLogger().Error(
-				"Execute auto analyze job failed",
-				zap.Stringer("job", job),
-				zap.Error(err),
+		wg.Run(func() {
+			err = job.Analyze(
+				r.statsHandle,
+				r.sysProcTracker,
 			)
+			if err != nil {
+				statslogutil.StatsLogger().Error(
+					"Execute auto analyze job failed",
+					zap.Stringer("job", job),
+					zap.Error(err),
+				)
+			}
+		})
+		cnt++
+		if cnt >= int(variable.AutoAnlayzeConcurrency.Load()) {
+			break
 		}
-		// Only analyze one table each time.
+	}
+	if cnt > 0 {
 		return true
 	}
 	statslogutil.SingletonStatsSamplerLogger().Info(

pkg/statistics/handle/autoanalyze/refresher/refresher_test.go

@@ -175,6 +175,7 @@ func TestPickOneTableAndAnalyzeByPriority(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
+	tk.MustExec("set global tidb_auto_analyze_concurrency=1")
 	tk.MustExec("create table t1 (a int, b int, index idx(a)) partition by range (a) (partition p0 values less than (2), partition p1 values less than (14))")
 	tk.MustExec("create table t2 (a int, b int, index idx(a)) partition by range (a) (partition p0 values less than (2), partition p1 values less than (14))")
 	tk.MustExec("insert into t1 values (1, 1), (2, 2), (3, 3)")
@@ -226,7 +227,7 @@ func TestPickOneTableAndAnalyzeByPriorityWithFailedAnalysis(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
-
+	tk.MustExec("set global tidb_auto_analyze_concurrency=1")
 	tk.MustExec("create table t1 (a int, b int, index idx(a)) partition by range (a) (partition p0 values less than (2), partition p1 values less than (4))")
 	tk.MustExec("create table t2 (a int, b int, index idx(a)) partition by range (a) (partition p0 values less than (2), partition p1 values less than (4))")
 	tk.MustExec("insert into t1 values (1, 1), (2, 2), (3, 3)")