fix: make new pq workable

Signed-off-by: Rustin170506 <29879298+Rustin170506@users.noreply.github.com>

pkg/infoschema/infoschema.go

@@ -308,6 +308,24 @@ func (is *infoSchema) TableByID(_ stdctx.Context, id int64) (val table.Table, ok
 	return slice[idx], true
 }
 
+func (is *infoSchema) SchemaNameByTableID(id int64) (val pmodel.CIStr, ok bool) {
+	if !tableIDIsValid(id) {
+		return
+	}
+
+	slice := is.sortedTablesBuckets[tableBucketIdx(id)]
+	idx := slice.searchTable(id)
+	if idx == -1 {
+		return
+	}
+	dbId := slice[idx].Meta().DBID
+	db, ok := is.SchemaByID(dbId)
+	if !ok {
+		return
+	}
+	return db.Name, true
+}
+
 // TableInfoByID implements InfoSchema.TableInfoByID
 func (is *infoSchema) TableInfoByID(id int64) (*model.TableInfo, bool) {
 	tbl, ok := is.TableByID(stdctx.Background(), id)

pkg/infoschema/infoschema_v2.go

@@ -646,6 +646,20 @@ func (is *infoschemaV2) TableByID(ctx context.Context, id int64) (val table.Tabl
 	return ret, true
 }
 
+func (is *infoschemaV2) SchemaNameByTableID(id int64) (val pmodel.CIStr, ok bool) {
+	if !tableIDIsValid(id) {
+		return
+	}
+
+	eq := func(a, b *tableItem) bool { return a.tableID == b.tableID }
+	itm, ok := search(is.byID, is.infoSchema.schemaMetaVersion, tableItem{tableID: id, schemaVersion: math.MaxInt64}, eq)
+	if !ok {
+		return pmodel.CIStr{}, false
+	}
+
+	return itm.dbName, true
+}
+
 // TableItem is exported from tableItem.
 type TableItem struct {
 	DBName    pmodel.CIStr

pkg/infoschema/interface.go

@@ -30,6 +30,7 @@ type InfoSchema interface {
 	context.MetaOnlyInfoSchema
 	TableByName(ctx stdctx.Context, schema, table pmodel.CIStr) (table.Table, error)
 	TableByID(ctx stdctx.Context, id int64) (table.Table, bool)
+	SchemaNameByTableID(id int64) (pmodel.CIStr, bool)
 	FindTableByPartitionID(partitionID int64) (table.Table, *model.DBInfo, *model.PartitionDefinition)
 	ListTablesWithSpecialAttribute(filter specialAttributeFilter) []tableInfoResult
 	base() *infoSchema

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

@@ -31,7 +31,6 @@ go_library(
         "//pkg/statistics/handle/util",
         "//pkg/util",
         "//pkg/util/intest",
-        "//pkg/util/logutil",
         "//pkg/util/timeutil",
         "@com_github_pkg_errors//:errors",
         "@com_github_tikv_client_go_v2//oracle",

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

@@ -85,7 +85,7 @@ func (pq *AnalysisPriorityQueueV2) init() error {
 			if err := pq.setAutoAnalysisTimeWindow(sctx); err != nil {
 				return err
 			}
-			if !pq.isWithinTimeWindow() {
+			if !pq.IsWithinTimeWindow() {
 				return nil
 			}
 			pruneMode := variable.PartitionPruneMode(sctx.GetSessionVars().PartitionPruneMode.Load())
@@ -108,7 +108,7 @@ func (pq *AnalysisPriorityQueueV2) init() error {
 			for _, db := range dbs {
 				// Sometimes the tables are too many. Auto-analyze will take too much time on it.
 				// so we need to check the available time.
-				if !pq.isWithinTimeWindow() {
+				if !pq.IsWithinTimeWindow() {
 					return nil
 				}
 				// Ignore the memory and system database.
@@ -208,7 +208,7 @@ func (pq *AnalysisPriorityQueueV2) init() error {
 }
 
 func (pq *AnalysisPriorityQueueV2) run() {
-	dmlFetchInterval := time.NewTicker(time.Minute * 5)
+	dmlFetchInterval := time.NewTicker(time.Second * 10)
 	defer dmlFetchInterval.Stop()
 	timeRefreshInterval := time.NewTicker(time.Minute * 10)
 	defer timeRefreshInterval.Stop()
@@ -232,7 +232,7 @@ func (pq *AnalysisPriorityQueueV2) fetchDMLUpdate() {
 		if err := pq.setAutoAnalysisTimeWindow(sctx); err != nil {
 			return err
 		}
-		if !pq.isWithinTimeWindow() {
+		if !pq.IsWithinTimeWindow() {
 			return nil
 		}
 		values := pq.statsHandle.Values()
@@ -283,7 +283,12 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 		pruneMode := variable.PartitionPruneMode(sctx.GetSessionVars().PartitionPruneMode.Load())
 
 		is := sctx.GetDomainInfoSchema().(infoschema.InfoSchema)
-		job, ok, _ := pq.inner.GetByKey(stats.PhysicalID)
+		schemaName, ok := is.SchemaNameByTableID(stats.PhysicalID)
+		if !ok {
+			statslogutil.StatsLogger().Warn("schema not found for table id", zap.Int64("table_id", stats.PhysicalID))
+			return nil
+		}
+		job, ok, _ = pq.inner.GetByKey(stats.PhysicalID)
 		if !ok {
 			tableInfo, ok := pq.statsHandle.TableInfoByID(is, stats.PhysicalID)
 			tableMeta := tableInfo.Meta()
@@ -294,7 +299,7 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 			partitionedTable := tableMeta.GetPartitionInfo()
 			if partitionedTable == nil {
 				job = jobFactory.CreateNonPartitionedTableAnalysisJob(
-					tableMeta.Name.O,
+					schemaName.O,
 					tableMeta,
 					stats,
 				)
@@ -309,7 +314,7 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 						}
 					}
 					job = jobFactory.CreateStaticPartitionAnalysisJob(
-						tableMeta.Name.O,
+						schemaName.O,
 						tableMeta,
 						partitionDef.ID,
 						partitionDef.Name.O,
@@ -318,7 +323,7 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 				} else {
 					partitionStats := GetPartitionStats(pq.statsHandle, tableMeta, partitionDefs)
 					job = jobFactory.CreateDynamicPartitionedTableAnalysisJob(
-						tableMeta.Name.O,
+						schemaName.O,
 						tableMeta,
 						stats,
 						partitionStats,
@@ -339,7 +344,7 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 				partitionDefs := partitionedTable.Definitions
 				partitionStats := GetPartitionStats(pq.statsHandle, tableMeta, partitionDefs)
 				job = jobFactory.CreateDynamicPartitionedTableAnalysisJob(
-					tableMeta.Name.O,
+					schemaName.O,
 					tableMeta,
 					stats,
 					partitionStats,
@@ -359,7 +364,7 @@ func (pq *AnalysisPriorityQueueV2) handleTableStats(stats *statistics.Table) {
 }
 
 func (pq *AnalysisPriorityQueueV2) refreshTime() {
-	if !pq.isWithinTimeWindow() {
+	if !pq.IsWithinTimeWindow() {
 		return
 	}
 	if err := statsutil.CallWithSCtx(pq.statsHandle.SPool(), func(sctx sessionctx.Context) error {
@@ -428,7 +433,7 @@ func (pq *AnalysisPriorityQueueV2) setAutoAnalysisTimeWindow(sctx sessionctx.Con
 	return nil
 }
 
-func (pq *AnalysisPriorityQueueV2) isWithinTimeWindow() bool {
+func (pq *AnalysisPriorityQueueV2) IsWithinTimeWindow() bool {
 	window := pq.autoAnalysisTimeWindow.Load().(*AutoAnalysisTimeWindow)
 	return window.IsWithinTimeWindow(time.Now())
 }

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

@@ -35,7 +35,7 @@ import (
 )
 
 // useAnalysisPriorityQueueV2 is a constant that controls whether to use the new priority queue implementation.
-const useAnalysisPriorityQueueV2 = true
+const useAnalysisPriorityQueueV2 = false
 
 // Refresher provides methods to refresh stats info.
 // NOTE: Refresher is not thread-safe.
@@ -90,8 +90,14 @@ func (r *Refresher) UpdateConcurrency() {
 
 // AnalyzeHighestPriorityTables picks tables with the highest priority and analyzes them.
 func (r *Refresher) AnalyzeHighestPriorityTables() bool {
-	if !r.autoAnalysisTimeWindow.IsWithinTimeWindow(time.Now()) {
-		return false
+	if useAnalysisPriorityQueueV2 {
+		if !r.jobsV2.IsWithinTimeWindow() {
+			return false
+		}
+	} else {
+		if !r.autoAnalysisTimeWindow.IsWithinTimeWindow(time.Now()) {
+			return false
+		}
 	}
 
 	se, err := r.statsHandle.SPool().Get()
@@ -129,6 +135,7 @@ func (r *Refresher) AnalyzeHighestPriorityTables() bool {
 			statslogutil.StatsLogger().Debug("Job already running, skipping", zap.Int64("tableID", job.GetTableID()))
 			continue
 		}
+		statslogutil.StatsLogger().Info("job is valid to analyze", zap.Stringer("job", job))
 		if valid, failReason := job.IsValidToAnalyze(sctx); !valid {
 			statslogutil.SingletonStatsSamplerLogger().Info(
 				"Table not ready for analysis",