stats: do not close the priority queue in DDL handler (#65271)

Author: 0xPoe

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

@@ -37,20 +37,21 @@ import (
 
 // HandleDDLEvent handles DDL events for the priority queue.
 func (pq *AnalysisPriorityQueue) HandleDDLEvent(_ context.Context, sctx sessionctx.Context, event *notifier.SchemaChangeEvent) error {
-	// Check if auto analyze is enabled.
-	if !vardef.RunAutoAnalyze.Load() {
-		// Close the priority queue if auto analyze is disabled.
-		// This ensures proper cleanup of the DDL notifier (mysql.tidb_ddl_notifier) and prevents the queue from remaining
-		// in an unknown state. When auto analyze is re-enabled, the priority queue can be properly re-initialized.
-		// NOTE: It is safe to call Close multiple times and it will get the lock internally.
-		pq.Close()
-		return nil
-	}
 	pq.syncFields.mu.Lock()
 	defer pq.syncFields.mu.Unlock()
 	// If the priority queue is not initialized, we should retry later.
 	if !pq.syncFields.initialized {
-		return notifier.ErrNotReadyRetryLater
+		// If auto analyze is enabled but the priority queue is not initialized,
+		// it means the priority queue initialization is not finished yet.
+		// So we should retry later.
+		if vardef.RunAutoAnalyze.Load() {
+			return notifier.ErrNotReadyRetryLater
+		}
+		// NOTE: If auto analyze is disabled and the priority queue is not initialized,
+		// we can just ignore the DDL events.
+		// SAFETY: Once auto analyze is enabled again, the priority queue will be initialized
+		// and it will recreate the jobs for all tables.
+		return nil
 	}
 
 	var err error

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

@@ -1108,7 +1108,7 @@ func TestCreateIndexUnderDDLAnalyzeEnabled(t *testing.T) {
 	require.Equal(t, columnInfo[0].Name.L, "c1")
 }
 
-func TestTurnOffAutoAnalyze(t *testing.T) {
+func TestTurnOffAutoAnalyzeAfterQueueInit(t *testing.T) {
 	store, do := testkit.CreateMockStoreAndDomain(t)
 	testKit := testkit.NewTestKit(t, store)
 	testKit.MustExec("use test")
@@ -1141,31 +1141,71 @@ func TestTurnOffAutoAnalyze(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, tableInfo.ID, job.GetTableID())
 
-	// Truncate table.
-	testKit.MustExec("truncate table t")
+	// Add a new index on column c1.
+	testKit.MustExec("alter table t add index idx1(c1)")
 
-	// Find the truncate table event.
-	truncateTableEvent := statstestutil.FindEvent(h.DDLEventCh(), model.ActionTruncateTable)
+	// Find the add index event.
+	addIndexEvent := statstestutil.FindEvent(h.DDLEventCh(), model.ActionAddIndex)
+	require.NotNil(t, addIndexEvent)
 
 	// Disable the auto analyze.
 	testKit.MustExec("set @@global.tidb_enable_auto_analyze = 0;")
 
-	// Handle the truncate table event.
-	err = statstestutil.HandleDDLEventWithTxn(h, truncateTableEvent)
+	// Handle the add index event.
+	err = statstestutil.HandleDDLEventWithTxn(h, addIndexEvent)
 	require.NoError(t, err)
 
-	sctx := testKit.Session().(sessionctx.Context)
-	// Handle the truncate table event in priority queue.
-	require.NoError(t, pq.HandleDDLEvent(ctx, sctx, truncateTableEvent))
-
-	// Because the auto analyze is turned off, the priority queue should be closed properly.
-	_, err = pq.IsEmptyForTest()
-	require.ErrorContains(t, err, "priority queue not initialized")
+	// Handle the add index event in priority queue.
+	require.NoError(t, statsutil.CallWithSCtx(h.SPool(), func(sctx sessionctx.Context) error {
+		return pq.HandleDDLEvent(ctx, sctx, addIndexEvent)
+	}, statsutil.FlagWrapTxn))
 
-	// The priority queue can be re-initialized after turning on the auto analyze.
-	// We manually mock it here.
-	require.NoError(t, pq.Initialize(ctx))
 	isEmpty, err = pq.IsEmptyForTest()
 	require.NoError(t, err)
-	require.True(t, isEmpty)
+	require.False(t, isEmpty)
+}
+
+func TestTurnOffAutoAnalyzeBeforeQueueInit(t *testing.T) {
+	store, do := testkit.CreateMockStoreAndDomain(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	enableAutoAnalyze(t, testKit)
+	testKit.MustExec("create table t (c1 int, c2 int, index idx(c1, c2))")
+	h := do.StatsHandle()
+	statstestutil.HandleNextDDLEventWithTxn(h)
+	// Insert some data.
+	testKit.MustExec("insert into t values (1,2),(2,2),(6,2),(11,2),(16,2)")
+	require.NoError(t, h.DumpStatsDeltaToKV(true))
+	require.NoError(t, h.Update(context.Background(), do.InfoSchema()))
+
+	statistics.AutoAnalyzeMinCnt = 0
+	defer func() {
+		statistics.AutoAnalyzeMinCnt = 1000
+	}()
+
+	// Disable the auto analyze.
+	testKit.MustExec("set @@global.tidb_enable_auto_analyze = 0;")
+
+	pq := priorityqueue.NewAnalysisPriorityQueue(h)
+	defer pq.Close()
+	ctx := context.Background()
+
+	// Add a new index on column c1.
+	testKit.MustExec("alter table t add index idx1(c1)")
+
+	// Find the add index event.
+	addIndexEvent := statstestutil.FindEvent(h.DDLEventCh(), model.ActionAddIndex)
+	require.NotNil(t, addIndexEvent)
+
+	// Handle the add index event.
+	err := statstestutil.HandleDDLEventWithTxn(h, addIndexEvent)
+	require.NoError(t, err)
+
+	// Handle the add index event in priority queue.
+	require.NoError(t, statsutil.CallWithSCtx(h.SPool(), func(sctx sessionctx.Context) error {
+		return pq.HandleDDLEvent(ctx, sctx, addIndexEvent)
+	}, statsutil.FlagWrapTxn))
+
+	_, err = pq.IsEmptyForTest()
+	require.ErrorContains(t, err, "priority queue not initialized")
 }

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

@@ -34,7 +34,7 @@ go_test(
         "worker_test.go",
     ],
     flaky = True,
-    shard_count = 11,
+    shard_count = 12,
     deps = [
         ":refresher",
         "//pkg/parser/ast",

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

@@ -92,14 +92,6 @@ func (r *Refresher) UpdateConcurrency() {
 // Usually, this is done by the caller through `util.CallWithSCtx`.
 func (r *Refresher) AnalyzeHighestPriorityTables(sctx sessionctx.Context) bool {
 	parameters := exec.GetAutoAnalyzeParameters(sctx)
-	err := r.setAutoAnalysisTimeWindow(parameters)
-	if err != nil {
-		statslogutil.StatsErrVerboseSampleLogger().Error("Set auto analyze time window failed", zap.Error(err))
-		return false
-	}
-	if !r.isWithinTimeWindow() {
-		return false
-	}
 	currentAutoAnalyzeRatio := exec.ParseAutoAnalyzeRatio(parameters[vardef.TiDBAutoAnalyzeRatio])
 	currentPruneMode := variable.PartitionPruneMode(sctx.GetSessionVars().PartitionPruneMode.Load())
 	if !r.jobs.IsInitialized() {
@@ -122,6 +114,20 @@ func (r *Refresher) AnalyzeHighestPriorityTables(sctx sessionctx.Context) bool {
 		}
 	}
 
+	// NOTE: This check must be done after initializing/rebuilding the queue.
+	// For example, if TiDB instances restart outside the time window, the queue will not be initialized.
+	// This means the DDL events are not being processed. This would prevent the DDL notifier from moving forward.
+	// Although this won't cause a correctness issue and it does not affect other handlers, it is still better to avoid this situation.
+	// We should make sure the queue is always initialized when the current instance is the owner.
+	err := r.setAutoAnalysisTimeWindow(parameters)
+	if err != nil {
+		statslogutil.StatsErrVerboseSampleLogger().Error("Set auto analyze time window failed", zap.Error(err))
+		return false
+	}
+	if !r.isWithinTimeWindow() {
+		return false
+	}
+
 	// Update the concurrency to the latest value.
 	r.UpdateConcurrency()
 	// Check remaining concurrency.

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

@@ -49,38 +49,72 @@ func TestTurnOffAndOnAutoAnalyze(t *testing.T) {
 	tk.MustExec("insert into t values (5, 4), (5, 5), (6, 6)")
 	require.NoError(t, handle.DumpStatsDeltaToKV(true))
 	require.NoError(t, handle.Update(context.Background(), dom.InfoSchema()))
-	sysProcTracker := dom.SysProcTracker()
-	r := refresher.NewRefresher(context.Background(), handle, sysProcTracker, dom.DDLNotifier())
-	defer r.Close()
+
+	// This would initialize the auto analyze queue.
+	require.True(t, handle.HandleAutoAnalyze())
 	require.NoError(t, util.CallWithSCtx(handle.SPool(), func(sctx sessionctx.Context) error {
-		require.True(t, handle.HandleAutoAnalyze())
 		return nil
 	}))
-	r.WaitAutoAnalyzeFinishedForTest()
 
+	// Disable auto analyze.
 	tk.MustExec("set @@global.tidb_enable_auto_analyze = 0;")
-
 	// Add a new index to generate DDL event.
 	tk.MustExec("alter table t add index idx_b(b);")
 	// Make sure the mysql.tidb_ddl_notifier is not empty.
 	rows := tk.MustQuery("select * from mysql.tidb_ddl_notifier").Rows()
 	require.Greater(t, len(rows), 0)
-	require.Eventually(t, func() bool {
-		return !r.IsQueueInitializedForTest()
-	}, time.Second*5, time.Millisecond*100)
 
 	// Make sure the mysql.tidb_ddl_notifier table is empty.
 	require.Eventually(t, func() bool {
 		rows := tk.MustQuery("select * from mysql.tidb_ddl_notifier").Rows()
 		return len(rows) == 0
 	}, time.Second*5, time.Millisecond*100)
 
-	// Enable auto analyze again to make sure the queue is re-initialized and handles the DDL event correctly.
+	// Make sure the table is added to the auto analyze queue.
+	snapshot, err := handle.GetPriorityQueueSnapshot()
+	require.NoError(t, err)
+	require.Equal(t, 1, len(snapshot.CurrentJobs), "The queue should have one job for table t with the new index idx_b")
+
+	// Enable auto analyze again to make sure the queue works.
 	tk.MustExec("set @@global.tidb_enable_auto_analyze = 1;")
+	require.True(t, handle.HandleAutoAnalyze())
+}
+
+func TestQueueInitializesOutsideTimeWindow(t *testing.T) {
+	statistics.AutoAnalyzeMinCnt = 0
+	defer func() {
+		statistics.AutoAnalyzeMinCnt = 1000
+	}()
+
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+
+	handle := dom.StatsHandle()
+	now := time.Now()
+	// Set the auto-analyze time window to a future time.
+	tk.MustExec("set @@global.tidb_auto_analyze_start_time = ?", now.Add(time.Hour).Format("15:04"))
+	tk.MustExec("set @@global.tidb_auto_analyze_end_time = ?", now.Add(2*time.Hour).Format("15:04"))
+
+	tk.MustExec("use test")
+	tk.MustExec("create table t (a int, b int, index idx(a))")
+	tk.MustExec("insert into t values (1, 1), (2, 2), (3, 3)")
+	require.NoError(t, handle.DumpStatsDeltaToKV(true))
+	require.NoError(t, handle.Update(context.Background(), dom.InfoSchema()))
+	tk.MustExec("analyze table t")
+
+	r := refresher.NewRefresher(context.Background(), handle, dom.SysProcTracker(), dom.DDLNotifier())
+	defer r.Close()
+
+	tk.MustExec("insert into t values (5, 4), (5, 5), (6, 6)")
+	require.NoError(t, handle.DumpStatsDeltaToKV(true))
+	require.NoError(t, handle.Update(context.Background(), dom.InfoSchema()))
+
 	require.NoError(t, util.CallWithSCtx(handle.SPool(), func(sctx sessionctx.Context) error {
-		require.True(t, handle.HandleAutoAnalyze())
+		require.False(t, r.AnalyzeHighestPriorityTables(sctx), "Because it's out of time window, no jobs should be executed")
 		return nil
 	}))
+	r.WaitAutoAnalyzeFinishedForTest()
+	require.Equal(t, 1, r.Len(), "The job queue should be initialized even if it is out of the time window")
 }
 
 func TestChangePruneMode(t *testing.T) {