WIP: statistics: add AnalysisPriorityQueueV2

pkg/owner/manager.go

@@ -500,16 +500,6 @@ func init() {
 	}
 }
 
-// DeleteLeader deletes the leader key.
-func DeleteLeader(ctx context.Context, cli *clientv3.Client, key string) error {
-	ownerKey, _, _, _, _, err := getOwnerInfo(ctx, ctx, cli, key)
-	if err != nil {
-		return errors.Trace(err)
-	}
-	_, err = cli.Delete(ctx, ownerKey)
-	return err
-}
-
 // AcquireDistributedLock creates a mutex with ETCD client, and returns a mutex release function.
 func AcquireDistributedLock(
 	ctx context.Context,
@@ -522,14 +512,7 @@ func AcquireDistributedLock(
 		return nil, err
 	}
 	mu := concurrency.NewMutex(se, key)
-	maxRetryCnt := 10
-	err = util2.RunWithRetry(maxRetryCnt, util2.RetryInterval, func() (bool, error) {
-		err = mu.Lock(ctx)
-		if err != nil {
-			return true, err
-		}
-		return false, nil
-	})
+	err = mu.Lock(ctx)
 	if err != nil {
 		err1 := se.Close()
 		if err1 != nil {

pkg/session/bootstrap.go

@@ -31,7 +31,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/bindinfo"
 	"github.com/pingcap/tidb/pkg/config"
-	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
@@ -63,9 +62,6 @@ import (
 	"go.uber.org/zap"
 )
 
-// bootstrapOwnerKey is the key used by ddl owner mutex during boostrap.
-var bootstrapOwnerKey = "/tidb/distributeDDLOwnerLock/"
-
 const (
 	// CreateUserTable is the SQL statement creates User table in system db.
 	// WARNING: There are some limitations on altering the schema of mysql.user table.
@@ -1421,48 +1417,6 @@ var (
 	SupportUpgradeHTTPOpVer int64 = version174
 )
 
-func acquireLock(s sessiontypes.Session) (func(), bool) {
-	dom := domain.GetDomain(s)
-	if dom == nil {
-		logutil.BgLogger().Warn("domain is nil")
-		return nil, false
-	}
-	cli := dom.GetEtcdClient()
-	if cli == nil {
-		logutil.BgLogger().Warn("etcd client is nil, force to acquire ddl owner lock")
-		// Special handling for test.
-		return func() {
-			// do nothing
-		}, true
-	}
-	releaseFn, err := owner.AcquireDistributedLock(context.Background(), cli, ddl.DDLOwnerKey, 10)
-	if err != nil {
-		return nil, false
-	}
-	return releaseFn, true
-}
-
-func forceToLeader(ctx context.Context, s sessiontypes.Session) error {
-	dom := domain.GetDomain(s)
-	for !dom.DDL().OwnerManager().IsOwner() {
-		ownerID, err := dom.DDL().OwnerManager().GetOwnerID(ctx)
-		if err != nil && (errors.ErrorEqual(err, concurrency.ErrElectionNoLeader) || strings.Contains(err.Error(), "no owner")) {
-			time.Sleep(50 * time.Millisecond)
-			continue
-		} else if err != nil {
-			logutil.BgLogger().Error("unexpected error", zap.Error(err))
-			return err
-		}
-		err = owner.DeleteLeader(ctx, dom.EtcdClient(), ddl.DDLOwnerKey)
-		if err != nil {
-			logutil.BgLogger().Error("unexpected error", zap.Error(err), zap.String("ownerID", ownerID))
-			return err
-		}
-		time.Sleep(50 * time.Millisecond)
-	}
-	return nil
-}
-
 func checkDistTask(s sessiontypes.Session, ver int64) {
 	if ver > version195 {
 		// since version195 we enable dist task by default, no need to check
@@ -1512,34 +1466,35 @@ func checkDistTask(s sessiontypes.Session, ver int64) {
 // upgrade function  will do some upgrade works, when the system is bootstrapped by low version TiDB server
 // For example, add new system variables into mysql.global_variables table.
 func upgrade(s sessiontypes.Session) {
-	// Do upgrade works then update bootstrap version.
-	isNull, err := InitMDLVariableForUpgrade(s.GetStore())
-	if err != nil {
-		logutil.BgLogger().Fatal("[upgrade] init metadata lock failed", zap.Error(err))
-	}
-
-	var ver int64
-	releaseFn, ok := acquireLock(s)
-	if !ok {
-		logutil.BgLogger().Fatal("[upgrade] get ddl owner distributed lock failed", zap.Error(err))
-	}
-	ver, err = getBootstrapVersion(s)
+	ver, err := getBootstrapVersion(s)
 	terror.MustNil(err)
 	if ver >= currentBootstrapVersion {
 		// It is already bootstrapped/upgraded by a higher version TiDB server.
-		releaseFn()
 		return
 	}
-	defer releaseFn()
-
-	err = forceToLeader(context.Background(), s)
-	if err != nil {
-		logutil.BgLogger().Fatal("[upgrade] force to owner failed", zap.Error(err))
-	}
 
 	checkDistTask(s, ver)
 	printClusterState(s, ver)
 
+	// Only upgrade from under version92 and this TiDB is not owner set.
+	// The owner in older tidb does not support concurrent DDL, we should add the internal DDL to job queue.
+	if ver < version92 {
+		useConcurrentDDL, err := checkOwnerVersion(context.Background(), domain.GetDomain(s))
+		if err != nil {
+			logutil.BgLogger().Fatal("[upgrade] upgrade failed", zap.Error(err))
+		}
+		if !useConcurrentDDL {
+			// Use another variable DDLForce2Queue but not EnableConcurrentDDL since in upgrade it may set global variable, the initial step will
+			// overwrite variable EnableConcurrentDDL.
+			variable.DDLForce2Queue.Store(true)
+		}
+	}
+	// Do upgrade works then update bootstrap version.
+	isNull, err := InitMDLVariableForUpgrade(s.GetStore())
+	if err != nil {
+		logutil.BgLogger().Fatal("[upgrade] init metadata lock failed", zap.Error(err))
+	}
+
 	// when upgrade from v6.4.0 or earlier, enables metadata lock automatically,
 	// but during upgrade we disable it.
 	if isNull {

pkg/statistics/handle/autoanalyze/internal/heap/heap.go

@@ -16,6 +16,7 @@
 // 1. Use the `errors` package from PingCAP.
 // 2. Use generics to define the `heapData` struct.
 // 3. Add a peak API.
+// 4. Add an IsEmpty API.
 
 package heap
 
@@ -221,6 +222,13 @@ func (h *Heap[K, V]) Peek() (V, error) {
 	return h.data.items[h.data.queue[0]].obj, nil
 }
 
+// IsEmpty returns true if the heap is empty.
+func (h *Heap[K, V]) IsEmpty() bool {
+	h.lock.RLock()
+	defer h.lock.RUnlock()
+	return len(h.data.queue) == 0
+}
+
 // Pop removes the top object from the heap and returns it.
 func (h *Heap[K, V]) Pop() (V, error) {
 	h.lock.Lock()

pkg/statistics/handle/autoanalyze/priorityqueue/BUILD.bazel

@@ -10,25 +10,29 @@ go_library(
         "job.go",
         "non_partitioned_table_analysis_job.go",
         "queue.go",
+        "queue2.go",
         "static_partitioned_table_analysis_job.go",
     ],
     importpath = "github.com/pingcap/tidb/pkg/statistics/handle/autoanalyze/priorityqueue",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/ddl/notifier",
         "//pkg/infoschema",
         "//pkg/meta/model",
         "//pkg/sessionctx",
         "//pkg/sessionctx/sysproctrack",
         "//pkg/sessionctx/variable",
         "//pkg/statistics",
         "//pkg/statistics/handle/autoanalyze/exec",
+        "//pkg/statistics/handle/autoanalyze/internal/heap",
         "//pkg/statistics/handle/lockstats",
         "//pkg/statistics/handle/logutil",
         "//pkg/statistics/handle/types",
         "//pkg/statistics/handle/util",
         "//pkg/util",
         "//pkg/util/intest",
         "//pkg/util/timeutil",
+        "@com_github_pkg_errors//:errors",
         "@com_github_tikv_client_go_v2//oracle",
         "@org_uber_go_zap//:zap",
     ],

pkg/statistics/handle/autoanalyze/priorityqueue/job.go

@@ -29,6 +29,7 @@ import (
 // NOTE: this is only used when the average analysis duration is not available.(No successful analysis before)
 const defaultFailedAnalysisWaitTime = 30 * time.Minute
 
+// analyzeType is the type of the analyze job.
 type analyzeType string
 
 // Indicators contains some indicators to evaluate the table priority.

pkg/statistics/handle/autoanalyze/priorityqueue/non_partitioned_table_analysis_job.go

@@ -126,7 +126,7 @@ func (j *NonPartitionedTableAnalysisJob) GetIndicators() Indicators {
 	return j.Indicators
 }
 
-// SetIndicators sets the indicators of the table.
+// SetIndicators sets the indicators of the job.
 func (j *NonPartitionedTableAnalysisJob) SetIndicators(indicators Indicators) {
 	j.Indicators = indicators
 }