statistics: fix targeted stats cache refresh after analyze

.agents/skills/tidb-test-guidelines/references/executor-case-map.md

@@ -258,7 +258,7 @@
 
 ### Tests
 - `pkg/executor/test/analyzetest/analyze_bench_test.go` - executor/test/analyzetest: Tests analyze partition.
-- `pkg/executor/test/analyzetest/analyze_test.go` - executor/test/analyzetest: Tests analyze partition.
+- `pkg/executor/test/analyzetest/analyze_test.go` - executor/test/analyzetest: Tests partition analyze execution and stats refresh.
 - `pkg/executor/test/analyzetest/main_test.go` - Configures default goleak settings and registers testdata.
 
 ## pkg/executor/test/analyzetest/columns

.agents/skills/tidb-test-guidelines/references/statistics-case-map.md

@@ -98,7 +98,7 @@
 
 ### Tests
 - `pkg/statistics/handle/globalstats/global_stats_internal_test.go` - statistics/handle/globalstats: Tests global stats internal.
-- `pkg/statistics/handle/globalstats/global_stats_test.go` - statistics/handle/globalstats: Tests show global stats with async merge global.
+- `pkg/statistics/handle/globalstats/global_stats_test.go` - statistics/handle/globalstats: Tests global stats merge and persistence behavior.
 - `pkg/statistics/handle/globalstats/main_test.go` - Configures default goleak settings and registers testdata.
 - `pkg/statistics/handle/globalstats/topn_bench_test.go` - statistics/handle/globalstats: Tests merge part top N to global top N with hists.
 - `pkg/statistics/handle/globalstats/topn_test.go` - statistics/handle/globalstats: Tests merge part top N to global top N without hists.
@@ -132,7 +132,7 @@
 
 ### Tests
 - `pkg/statistics/handle/handletest/statstest/main_test.go` - Configures default goleak settings and registers testdata.
-- `pkg/statistics/handle/handletest/statstest/stats_test.go` - statistics/handle/handletest: Tests stats cache process.
+- `pkg/statistics/handle/handletest/statstest/stats_test.go` - statistics/handle/handletest: Tests full and targeted stats cache updates.
 
 ## pkg/statistics/handle/lockstats
 

pkg/executor/analyze.go

@@ -324,12 +324,16 @@ func (e *AnalyzeExec) Next(ctx context.Context, _ *chunk.Chunk) (err error) {
 	if len(tasks) == 0 {
 		return nil
 	}
-	tableAndPartitionIDs := make([]int64, 0, len(tasks))
+	// Analyze results are persisted under physical statistics IDs. Logical
+	// partition-table metadata is refreshed separately unless a global stats
+	// merge persists a new histogram that requires a full reload.
+	statsIDsToRefresh := make([]int64, 0, len(tasks))
+	logicalTableMetaIDsToRefresh := make(map[int64]struct{})
 	for _, task := range tasks {
 		tableID := getTableIDFromTask(task)
-		tableAndPartitionIDs = append(tableAndPartitionIDs, tableID.TableID)
+		statsIDsToRefresh = append(statsIDsToRefresh, tableID.GetStatisticsID())
 		if tableID.IsPartitionTable() {
-			tableAndPartitionIDs = append(tableAndPartitionIDs, tableID.PartitionID)
+			logicalTableMetaIDsToRefresh[tableID.TableID] = struct{}{}
 		}
 	}
 
@@ -423,9 +427,10 @@ TASKLOOP:
 	})
 	// If we enabled dynamic prune mode, then we need to generate global stats here for partition tables.
 	if needGlobalStats {
-		err = e.handleGlobalStats(statsHandle, globalStatsMap)
-		if err != nil {
-			return err
+		globalStatsIDs := e.handleGlobalStats(statsHandle, globalStatsMap)
+		statsIDsToRefresh = append(statsIDsToRefresh, globalStatsIDs...)
+		for _, tableID := range globalStatsIDs {
+			delete(logicalTableMetaIDsToRefresh, tableID)
 		}
 	}
 
@@ -446,7 +451,14 @@ TASKLOOP:
 	if err != nil {
 		sessionVars.StmtCtx.AppendWarning(err)
 	}
-	return statsHandle.Update(ctx, infoSchema, tableAndPartitionIDs...)
+	if err := statsHandle.Update(ctx, infoSchema, statsIDsToRefresh...); err != nil {
+		return err
+	}
+	logicalTableMetaIDs := make([]int64, 0, len(logicalTableMetaIDsToRefresh))
+	for tableID := range logicalTableMetaIDsToRefresh {
+		logicalTableMetaIDs = append(logicalTableMetaIDs, tableID)
+	}
+	return statsHandle.UpdateTableStatsMeta(ctx, infoSchema, logicalTableMetaIDs...)
 }
 
 func (e *AnalyzeExec) waitFinish(ctx context.Context, g *errgroup.Group, resultsCh chan *statistics.AnalyzeResults) error {

pkg/executor/analyze_global_stats.go

@@ -37,15 +37,17 @@ type globalStatsKey struct {
 // The meaning of value in map is some additional information needed to build global-level stats.
 type globalStatsMap map[globalStatsKey]statstypes.GlobalStatsInfo
 
-func (e *AnalyzeExec) handleGlobalStats(statsHandle *handle.Handle, globalStatsMap globalStatsMap) error {
+func (e *AnalyzeExec) handleGlobalStats(statsHandle *handle.Handle, globalStatsMap globalStatsMap) []int64 {
 	globalStatsTableIDs := make(map[int64]struct{}, len(globalStatsMap))
 	for globalStatsID := range globalStatsMap {
 		globalStatsTableIDs[globalStatsID.tableID] = struct{}{}
 	}
 
 	tableIDs := make(map[int64]struct{}, len(globalStatsTableIDs))
+	updatedTableIDs := make([]int64, 0, len(globalStatsTableIDs))
 	for tableID := range globalStatsTableIDs {
 		tableIDs[tableID] = struct{}{}
+		tableUpdated := false
 		for globalStatsID, info := range globalStatsMap {
 			if globalStatsID.tableID != tableID {
 				continue
@@ -58,14 +60,14 @@ func (e *AnalyzeExec) handleGlobalStats(statsHandle *handle.Handle, globalStatsM
 			AddNewAnalyzeJob(e.Ctx(), job)
 			statsHandle.StartAnalyzeJob(job)
 
-			mergeStatsErr := func() error {
+			globalStatsUpdated, mergeStatsErr := func() (bool, error) {
 				globalOpts := e.opts
 				if e.OptionsMap != nil {
 					if v2Options, ok := e.OptionsMap[globalStatsID.tableID]; ok {
 						globalOpts = v2Options.FilledOpts
 					}
 				}
-				err := statsHandle.MergePartitionStats2GlobalStatsByTableID(
+				updated, err := statsHandle.MergePartitionStats2GlobalStatsByTableID(
 					e.Ctx(),
 					globalOpts, e.Ctx().GetInfoSchema().(infoschema.InfoSchema),
 					&info,
@@ -75,10 +77,16 @@ func (e *AnalyzeExec) handleGlobalStats(statsHandle *handle.Handle, globalStatsM
 					logutil.ErrVerboseLogger().Warn("merge global stats failed",
 						zap.String("info", job.JobInfo), zap.Error(err), zap.Int64("tableID", tableID))
 				}
-				return err
+				return updated, err
 			}()
+			// Column and index stats are persisted separately. Refresh the table
+			// when any merge wrote data, even if another merge failed.
+			tableUpdated = tableUpdated || globalStatsUpdated
 			statsHandle.FinishAnalyzeJob(job, mergeStatsErr, statistics.GlobalStatsMergeJob)
 		}
+		if tableUpdated {
+			updatedTableIDs = append(updatedTableIDs, tableID)
+		}
 	}
 
 	for tableID := range tableIDs {
@@ -88,7 +96,7 @@ func (e *AnalyzeExec) handleGlobalStats(statsHandle *handle.Handle, globalStatsM
 		}
 	}
 
-	return nil
+	return updatedTableIDs
 }
 
 func (e *AnalyzeExec) newAnalyzeHandleGlobalStatsJob(key globalStatsKey) *statistics.AnalyzeJob {

pkg/executor/test/analyzetest/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 40,
+    shard_count = 41,
     deps = [
         "//pkg/config",
         "//pkg/config/kerneltype",

pkg/executor/test/analyzetest/analyze_test.go

@@ -129,6 +129,114 @@ PARTITION BY RANGE ( a ) (
 	})
 }
 
+// TestAnalyzeRefreshesOnlyPersistedStats verifies that partition ANALYZE only
+// refreshes physical or global stats IDs whose persistence path wrote data.
+// Each case caches a pseudo global entry behind a higher cache-wide version,
+// analyzes one partition, and checks both the global entry's merge state and
+// its logical-table row count.
+func TestAnalyzeRefreshesOnlyPersistedStats(t *testing.T) {
+	testCases := []struct {
+		name                     string
+		partitionPruneMode       string
+		makeGlobalStatsMergeFail bool
+		expectGlobalStatsRefresh bool
+	}{
+		{
+			name:               "static partition analyze",
+			partitionPruneMode: "static",
+		},
+		{
+			name:                     "successful dynamic global stats merge",
+			partitionPruneMode:       "dynamic",
+			expectGlobalStatsRefresh: true,
+		},
+		{
+			name:                     "failed dynamic global stats merge",
+			partitionPruneMode:       "dynamic",
+			makeGlobalStatsMergeFail: true,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			// Set up partition stats and a persisted global row, then arrange
+			// either no global merge, a successful merge, or a pre-write failure.
+			store, dom := testkit.CreateMockStoreAndDomain(t)
+			tk := testkit.NewTestKit(t, store)
+			tk.MustExec("use test")
+			tk.MustExec("set @@session.tidb_analyze_version = 2")
+			tk.MustExec("set @@session.tidb_partition_prune_mode = 'dynamic'")
+			tk.MustExec("set @@session.tidb_enable_async_merge_global_stats = ON")
+			tk.MustExec(`create table t(a int) partition by range(a) (
+				partition p0 values less than (10),
+				partition p1 values less than (20)
+			)`)
+			tk.MustExec("insert into t values (1), (11)")
+			analyzehelper.TriggerPredicateColumnsCollection(t, tk, store, "t", "a")
+			tk.MustExec("analyze table t")
+			if testCase.makeGlobalStatsMergeFail {
+				// p1 has no stats for the newly added predicate column, so
+				// merging p0 with missing-partition skipping disabled fails
+				// before any global row is written.
+				tk.MustExec("alter table t add column b int")
+				analyzehelper.TriggerPredicateColumnsCollection(t, tk, store, "t", "a", "b")
+				tk.MustExec("set @@session.tidb_skip_missing_partition_stats = 0")
+				tk.MustExec("insert into t values (2, 2), (3, 3)")
+			} else {
+				tk.MustExec("insert into t values (2), (3)")
+			}
+			// Upstream flushes pending deltas before building ANALYZE tasks.
+			// Flush first, then lower the global row version so that pre-flush
+			// cannot accidentally remove the intended stale-row setup.
+			tk.MustExec("flush stats_delta *.*")
+
+			table, err := dom.InfoSchema().TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t"))
+			require.NoError(t, err)
+			tableInfo := table.Meta()
+			partitionInfo := tableInfo.GetPartitionInfo()
+			require.NotNil(t, partitionInfo)
+			p0ID := partitionInfo.Definitions[0].ID
+
+			const (
+				staleGlobalVersion  uint64 = 100
+				cacheWideMaxVersion uint64 = 200
+			)
+			tk.MustExec("update mysql.stats_meta set version = ? where table_id = ?", staleGlobalVersion, tableInfo.ID)
+			handle := dom.StatsHandle()
+			handle.Clear()
+			globalStats := handle.GetPhysicalTableStats(tableInfo.ID, tableInfo)
+			require.True(t, globalStats.Pseudo)
+			handle.Put(tableInfo.ID+1000, &statistics.Table{
+				HistColl:              statistics.HistColl{PhysicalID: tableInfo.ID + 1000},
+				Version:               cacheWideMaxVersion,
+				ColAndIdxExistenceMap: statistics.NewColAndIndexExistenceMapWithoutSize(),
+			})
+			handle.WaitForAsyncUpdates()
+			require.Equal(t, cacheWideMaxVersion, handle.MaxTableStatsVersion())
+
+			tk.MustExec("set @@session.tidb_partition_prune_mode = ?", testCase.partitionPruneMode)
+			tk.MustExec("analyze table t partition p0")
+			if testCase.makeGlobalStatsMergeFail {
+				// Global merge errors are warnings rather than fatal ANALYZE
+				// errors; assert the failure path ran before checking the cache.
+				warnings := fmt.Sprint(tk.MustQuery("show warnings").Rows())
+				require.Contains(t, warnings, "Build global-level stats failed")
+			}
+
+			partitionStats := handle.GetPhysicalTableStats(p0ID, tableInfo)
+			require.False(t, partitionStats.Pseudo)
+			globalStats = handle.GetPhysicalTableStats(tableInfo.ID, tableInfo)
+			if testCase.expectGlobalStatsRefresh {
+				require.False(t, globalStats.Pseudo)
+			} else {
+				require.True(t, globalStats.Pseudo)
+			}
+			require.Equal(t, int64(4), globalStats.RealtimeCount)
+			require.Equal(t, cacheWideMaxVersion, handle.MaxTableStatsVersion())
+		})
+	}
+}
+
 func TestAnalyzeReplicaReadFollower(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)