statistics: handle deleted tables correctly in the PQ

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

@@ -57,7 +57,7 @@ func (f *AnalysisJobFactory) CreateNonPartitionedTableAnalysisJob(
 	tblInfo *model.TableInfo,
 	tblStats *statistics.Table,
 ) AnalysisJob {
-	if !tblStats.IsEligibleForAnalysis() {
+	if tblStats == nil || !tblStats.IsEligibleForAnalysis() {
 		return nil
 	}
 
@@ -92,7 +92,7 @@ func (f *AnalysisJobFactory) CreateStaticPartitionAnalysisJob(
 	partitionID int64,
 	partitionStats *statistics.Table,
 ) AnalysisJob {
-	if !partitionStats.IsEligibleForAnalysis() {
+	if partitionStats == nil || !partitionStats.IsEligibleForAnalysis() {
 		return nil
 	}
 
@@ -128,7 +128,7 @@ func (f *AnalysisJobFactory) CreateDynamicPartitionedTableAnalysisJob(
 	globalTblStats *statistics.Table,
 	partitionStats map[PartitionIDAndName]*statistics.Table,
 ) AnalysisJob {
-	if !globalTblStats.IsEligibleForAnalysis() {
+	if globalTblStats == nil || !globalTblStats.IsEligibleForAnalysis() {
 		return nil
 	}
 

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

@@ -644,14 +644,16 @@ func (pq *AnalysisPriorityQueue) RefreshLastAnalysisDuration() {
 					zap.Int64("tableID", job.GetTableID()),
 					zap.String("job", job.String()),
 				)
-				// TODO: Remove this after handling the DDL event.
+				// Delete the job from the queue since its table is missing. This is a safeguard -
+				// DDL events should have already cleaned up jobs for dropped tables.
 				err := pq.syncFields.inner.delete(job)
 				if err != nil {
 					statslogutil.StatsLogger().Error("Failed to delete job from priority queue",
 						zap.Error(err),
 						zap.String("job", job.String()),
 					)
 				}
+				continue
 			}
 			indicators.LastAnalysisDuration = jobFactory.GetTableLastAnalyzeDuration(tableStats)
 			job.SetIndicators(indicators)
@@ -852,9 +854,9 @@ func (pq *AnalysisPriorityQueue) Close() {
 	pq.syncFields.initialized = false
 	// The rest fields will be reset when the priority queue is initialized.
 	// But we do it here for double safety.
-	pq.syncFields.inner = nil
-	pq.syncFields.runningJobs = nil
-	pq.syncFields.mustRetryJobs = nil
+	pq.syncFields.inner = newHeap()
+	pq.syncFields.runningJobs = make(map[int64]struct{})
+	pq.syncFields.mustRetryJobs = make(map[int64]struct{})
 	pq.syncFields.lastDMLUpdateFetchTimestamp = 0
 	pq.syncFields.cancel = nil
 }

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

@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/tidb/pkg/meta/model"
 	pmodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/statistics"
@@ -608,3 +609,47 @@ func TestPQCanBeClosedAndReInitialized(t *testing.T) {
 	// Check if the priority queue is initialized.
 	require.True(t, pq.IsInitialized())
 }
+
+func TestPQHandlesTableDeletionGracefully(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	handle := dom.StatsHandle()
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1 (a int)")
+	tk.MustExec("insert into t1 values (1)")
+	statistics.AutoAnalyzeMinCnt = 0
+	defer func() {
+		statistics.AutoAnalyzeMinCnt = 1000
+	}()
+
+	ctx := context.Background()
+	require.NoError(t, handle.DumpStatsDeltaToKV(true))
+	require.NoError(t, handle.Update(ctx, dom.InfoSchema()))
+	pq := priorityqueue.NewAnalysisPriorityQueue(handle)
+	defer pq.Close()
+	require.NoError(t, pq.Initialize())
+
+	// Check the priority queue is not empty.
+	l, err := pq.Len()
+	require.NoError(t, err)
+	require.NotEqual(t, 0, l)
+
+	tbl, err := dom.InfoSchema().TableByName(ctx, pmodel.NewCIStr("test"), pmodel.NewCIStr("t1"))
+	require.NoError(t, err)
+
+	// Drop the table and mock the table stats is removed from the cache.
+	tk.MustExec("drop table t1")
+	deleteEvent := findEvent(handle.DDLEventCh(), model.ActionDropTable)
+	require.NotNil(t, deleteEvent)
+	require.NoError(t, handle.HandleDDLEvent(deleteEvent))
+	require.NoError(t, handle.Update(ctx, dom.InfoSchema()))
+
+	// Make sure handle.Get() returns false.
+	_, ok := handle.Get(tbl.Meta().ID)
+	require.False(t, ok)
+
+	require.NotPanics(t, func() {
+		pq.RefreshLastAnalysisDuration()
+	})
+}

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

@@ -261,6 +261,7 @@ func (*Refresher) OnBecomeOwner() {
 
 // OnRetireOwner is used to handle the event when the current TiDB instance retires from being the stats owner.
 func (r *Refresher) OnRetireOwner() {
-	// Stop the worker and close the queue.
+	// Theoretically we should stop the worker here, but stopping analysis jobs can be time-consuming.
+	// To avoid blocking etcd leader re-election, we only close the priority queue.
 	r.jobs.Close()
 }