From 0842d79a33827fda6ef158b8d75dcc4bf76d602b Mon Sep 17 00:00:00 2001 From: wlwilliamx Date: Thu, 16 Jul 2026 14:43:24 +0800 Subject: [PATCH 1/2] statistics: fix targeted stats cache refresh after analyze Targeted stats refreshes reused the cache-wide version lower bound. An unrelated table could therefore move the watermark past the analyzed table and leave pseudo stats in memory after ANALYZE. Reload explicitly requested IDs without that lower bound and only refresh physical or global IDs whose analyze results were persisted. Preserve the first global stats write error while tracking partial writes. --- .../references/executor-case-map.md | 2 +- .../references/statistics-case-map.md | 4 +- pkg/executor/analyze.go | 16 +-- pkg/executor/analyze_global_stats.go | 18 ++- pkg/executor/test/analyzetest/BUILD.bazel | 2 +- pkg/executor/test/analyzetest/analyze_test.go | 106 ++++++++++++++++++ pkg/statistics/handle/cache/statscache.go | 13 ++- pkg/statistics/handle/globalstats/BUILD.bazel | 2 +- .../handle/globalstats/global_stats.go | 24 ++-- .../handle/globalstats/global_stats_test.go | 46 ++++++++ .../handle/handletest/statstest/stats_test.go | 44 ++++++-- pkg/statistics/handle/types/interfaces.go | 3 +- 12 files changed, 239 insertions(+), 41 deletions(-) diff --git a/.agents/skills/tidb-test-guidelines/references/executor-case-map.md b/.agents/skills/tidb-test-guidelines/references/executor-case-map.md index 1a14fcc5a97cd..9a36825b2a057 100644 --- a/.agents/skills/tidb-test-guidelines/references/executor-case-map.md +++ b/.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 diff --git a/.agents/skills/tidb-test-guidelines/references/statistics-case-map.md b/.agents/skills/tidb-test-guidelines/references/statistics-case-map.md index f71c76a2b8e3a..4e7f7f0a42392 100644 --- a/.agents/skills/tidb-test-guidelines/references/statistics-case-map.md +++ b/.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 diff --git a/pkg/executor/analyze.go b/pkg/executor/analyze.go index 30addfe0a457b..8627a73300a0b 100644 --- a/pkg/executor/analyze.go +++ b/pkg/executor/analyze.go @@ -324,13 +324,12 @@ 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. The logical + // table ID is added later only when global stats are actually persisted. + statsIDsToRefresh := make([]int64, 0, len(tasks)) for _, task := range tasks { tableID := getTableIDFromTask(task) - tableAndPartitionIDs = append(tableAndPartitionIDs, tableID.TableID) - if tableID.IsPartitionTable() { - tableAndPartitionIDs = append(tableAndPartitionIDs, tableID.PartitionID) - } + statsIDsToRefresh = append(statsIDsToRefresh, tableID.GetStatisticsID()) } // Get the min number of goroutines for parallel execution. @@ -423,10 +422,7 @@ 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 - } + statsIDsToRefresh = append(statsIDsToRefresh, e.handleGlobalStats(statsHandle, globalStatsMap)...) } if intest.EnableInternalCheck { @@ -446,7 +442,7 @@ TASKLOOP: if err != nil { sessionVars.StmtCtx.AppendWarning(err) } - return statsHandle.Update(ctx, infoSchema, tableAndPartitionIDs...) + return statsHandle.Update(ctx, infoSchema, statsIDsToRefresh...) } func (e *AnalyzeExec) waitFinish(ctx context.Context, g *errgroup.Group, resultsCh chan *statistics.AnalyzeResults) error { diff --git a/pkg/executor/analyze_global_stats.go b/pkg/executor/analyze_global_stats.go index dc9846f3602c5..3a94732f0c7eb 100644 --- a/pkg/executor/analyze_global_stats.go +++ b/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 { diff --git a/pkg/executor/test/analyzetest/BUILD.bazel b/pkg/executor/test/analyzetest/BUILD.bazel index 460e8dfc356d8..720af68239c8b 100644 --- a/pkg/executor/test/analyzetest/BUILD.bazel +++ b/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", diff --git a/pkg/executor/test/analyzetest/analyze_test.go b/pkg/executor/test/analyzetest/analyze_test.go index 770d0f88f0fbf..b4f9293f3ef71 100644 --- a/pkg/executor/test/analyzetest/analyze_test.go +++ b/pkg/executor/test/analyzetest/analyze_test.go @@ -129,6 +129,112 @@ 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 the global entry against the merge result. +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, cacheWideMaxVersion, handle.MaxTableStatsVersion()) + }) + } +} + func TestAnalyzeReplicaReadFollower(t *testing.T) { store := testkit.CreateMockStore(t) tk := testkit.NewTestKit(t, store) diff --git a/pkg/statistics/handle/cache/statscache.go b/pkg/statistics/handle/cache/statscache.go index fd98ba1eb884e..0c58782f6821a 100644 --- a/pkg/statistics/handle/cache/statscache.go +++ b/pkg/statistics/handle/cache/statscache.go @@ -126,15 +126,14 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t dur := time.Since(start) tidbmetrics.StatsDeltaLoadHistogram.Observe(dur.Seconds()) }() - lastVersion := s.GetNextCheckVersionWithOffset() var ( skipMoveForwardStatsCache bool rows []chunk.Row err error ) if err := util.CallWithSCtx(s.statsHandle.SPool(), func(sctx sessionctx.Context) error { - query := "SELECT version, table_id, modify_count, count, snapshot, last_stats_histograms_version from mysql.stats_meta where version > %? " - args := []any{lastVersion} + query := "SELECT version, table_id, modify_count, count, snapshot, last_stats_histograms_version from mysql.stats_meta " + args := make([]any, 0, 2) if onlyForAnalyzedTables { // When updating specific tables, we skip incrementing the max stats version to avoid missing @@ -148,8 +147,14 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t for _, tableID := range tableAndPartitionIDs { tableStringIDs = append(tableStringIDs, strconv.FormatInt(tableID, 10)) } - query += "and table_id in (%?) " + // A targeted update follows ANALYZE and must reload each persisted result even + // when an unrelated table has advanced the cache-wide version beyond it. + query += "where table_id in (%?) " args = append(args, tableStringIDs) + } else { + lastVersion := s.GetNextCheckVersionWithOffset() + query += "where version > %? " + args = append(args, lastVersion) } query += "order by version" rows, _, err = util.ExecRows(sctx, query, args...) diff --git a/pkg/statistics/handle/globalstats/BUILD.bazel b/pkg/statistics/handle/globalstats/BUILD.bazel index fb61e9746f29b..5a3accc95a157 100644 --- a/pkg/statistics/handle/globalstats/BUILD.bazel +++ b/pkg/statistics/handle/globalstats/BUILD.bazel @@ -45,7 +45,7 @@ go_test( ], embed = [":globalstats"], flaky = True, - shard_count = 28, + shard_count = 29, deps = [ "//pkg/domain", "//pkg/kv", diff --git a/pkg/statistics/handle/globalstats/global_stats.go b/pkg/statistics/handle/globalstats/global_stats.go index c8d735bdbbd63..fe61253781ca2 100644 --- a/pkg/statistics/handle/globalstats/global_stats.go +++ b/pkg/statistics/handle/globalstats/global_stats.go @@ -48,18 +48,19 @@ func NewStatsGlobal(statsHandler statstypes.StatsHandle) statstypes.StatsGlobal } // MergePartitionStats2GlobalStatsByTableID merge the partition-level stats to global-level stats based on the tableID. +// The returned boolean reports whether at least one global stats record was persisted. func (sg *statsGlobalImpl) MergePartitionStats2GlobalStatsByTableID(sc sessionctx.Context, opts map[ast.AnalyzeOptionType]uint64, is infoschema.InfoSchema, info *statstypes.GlobalStatsInfo, physicalID int64, -) (err error) { +) (updated bool, err error) { globalStats, err := MergePartitionStats2GlobalStatsByTableID(sc, sg.statsHandler, opts, is, physicalID, info.IsIndex == 1, info.HistIDs) if err != nil { if types.ErrPartitionStatsMissing.Equal(err) || types.ErrPartitionColumnStatsMissing.Equal(err) { // When we find some partition-level stats are missing, we need to report warning. sc.GetSessionVars().StmtCtx.AppendWarning(err) } - return err + return false, err } return WriteGlobalStatsToStorage(sg.statsHandler, globalStats, info, physicalID) } @@ -359,8 +360,10 @@ func blockingMergePartitionStats2GlobalStats( return } -// WriteGlobalStatsToStorage is to write global stats to storage -func WriteGlobalStatsToStorage(statsHandle statstypes.StatsHandle, globalStats *GlobalStats, info *statstypes.GlobalStatsInfo, gid int64) (err error) { +// WriteGlobalStatsToStorage writes global stats to storage. It continues after +// individual write errors, reports whether any record was persisted, and returns +// the first write error. +func WriteGlobalStatsToStorage(statsHandle statstypes.StatsHandle, globalStats *GlobalStats, info *statstypes.GlobalStatsInfo, gid int64) (updated bool, firstErr error) { // Dump global-level stats to kv. for i := range globalStats.Num { hg, cms, topN := globalStats.Hg[i], globalStats.Cms[i], globalStats.TopN[i] @@ -369,7 +372,7 @@ func WriteGlobalStatsToStorage(statsHandle statstypes.StatsHandle, globalStats * continue } // fms for global stats doesn't need to dump to kv. - err = statsHandle.SaveColOrIdxStatsToStorage(gid, + saveErr := statsHandle.SaveColOrIdxStatsToStorage(gid, globalStats.Count, globalStats.ModifyCount, info.IsIndex, @@ -380,10 +383,15 @@ func WriteGlobalStatsToStorage(statsHandle statstypes.StatsHandle, globalStats * true, util.StatsMetaHistorySourceAnalyze, ) - if err != nil { + if saveErr != nil { + if firstErr == nil { + firstErr = saveErr + } statslogutil.StatsLogger().Warn("save global-level stats to storage failed", - zap.Int64("histID", hg.ID), zap.Error(err), zap.Int64("tableID", gid)) + zap.Int64("histID", hg.ID), zap.Error(saveErr), zap.Int64("tableID", gid)) + } else { + updated = true } } - return err + return updated, firstErr } diff --git a/pkg/statistics/handle/globalstats/global_stats_test.go b/pkg/statistics/handle/globalstats/global_stats_test.go index e5282aa4f414f..c44c34550a8f8 100644 --- a/pkg/statistics/handle/globalstats/global_stats_test.go +++ b/pkg/statistics/handle/globalstats/global_stats_test.go @@ -16,6 +16,7 @@ package globalstats_test import ( "context" + "errors" "fmt" "testing" "time" @@ -25,7 +26,9 @@ import ( "github.com/pingcap/tidb/pkg/planner/core" "github.com/pingcap/tidb/pkg/session" "github.com/pingcap/tidb/pkg/sessionctx" + "github.com/pingcap/tidb/pkg/statistics" statstestutil "github.com/pingcap/tidb/pkg/statistics/handle/ddl/testutil" + "github.com/pingcap/tidb/pkg/statistics/handle/globalstats" "github.com/pingcap/tidb/pkg/statistics/handle/types" "github.com/pingcap/tidb/pkg/testkit" "github.com/pingcap/tidb/pkg/testkit/testfailpoint" @@ -34,6 +37,49 @@ import ( const asyncMergeWarn = "Warning 1105 The 'tidb_enable_async_merge_global_stats' variable will always be enabled in a future release; changing it is discouraged." +type saveResultStatsHandle struct { + // Embed methods outside the persistence operation exercised by this test. + types.StatsHandle + saveResults []error + saveCalls int +} + +func (h *saveResultStatsHandle) SaveColOrIdxStatsToStorage( + _ int64, + _, _ int64, + _ int, + _ *statistics.Histogram, + _ *statistics.CMSketch, + _ *statistics.TopN, + _ int, + _ bool, + _ string, +) error { + result := h.saveResults[h.saveCalls] + h.saveCalls++ + return result +} + +// TestWriteGlobalStatsToStoragePreservesFirstError verifies that persistence +// continues after a failure without hiding it. It injects a failure followed by +// success and expects both a partial-update result and the first error. +func TestWriteGlobalStatsToStoragePreservesFirstError(t *testing.T) { + firstErr := errors.New("first histogram write failed") + handle := &saveResultStatsHandle{saveResults: []error{firstErr, nil}} + globalStats := &globalstats.GlobalStats{ + Hg: []*statistics.Histogram{{ID: 1}, {ID: 2}}, + Cms: make([]*statistics.CMSketch, 2), + TopN: make([]*statistics.TopN, 2), + Num: 2, + } + + updated, err := globalstats.WriteGlobalStatsToStorage(handle, globalStats, &types.GlobalStatsInfo{}, 42) + + require.True(t, updated) + require.ErrorIs(t, err, firstErr) + require.Equal(t, 2, handle.saveCalls) +} + func TestShowGlobalStatsWithAsyncMergeGlobal(t *testing.T) { testShowGlobalStats(t, true) } diff --git a/pkg/statistics/handle/handletest/statstest/stats_test.go b/pkg/statistics/handle/handletest/statstest/stats_test.go index f8c532b84dea4..567f219e4d73e 100644 --- a/pkg/statistics/handle/handletest/statstest/stats_test.go +++ b/pkg/statistics/handle/handletest/statstest/stats_test.go @@ -168,13 +168,15 @@ func checkPredicateColumnStats(t *testing.T, tableStats *statistics.Table, table }) } +// TestStatsCacheProcess verifies that targeted updates refresh analyzed tables +// without advancing the cache-wide version, while full updates advance it. func TestStatsCacheProcess(t *testing.T) { store, dom := testkit.CreateMockStoreAndDomain(t) - testKit := testkit.NewTestKit(t, store) - testKit.MustExec("use test") - testKit.MustExec("create table t (c1 int, c2 int)") - testKit.MustExec("insert into t values(1, 2)") - analyzehelper.TriggerPredicateColumnsCollection(t, testKit, store, "t", "c1", "c2") + tk := testkit.NewTestKit(t, store) + tk.MustExec("use test") + tk.MustExec("create table t (c1 int, c2 int)") + tk.MustExec("insert into t values(1, 2)") + analyzehelper.TriggerPredicateColumnsCollection(t, tk, store, "t", "c1", "c2") do := dom is := do.InfoSchema() tbl, err := is.TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t")) @@ -184,7 +186,7 @@ func TestStatsCacheProcess(t *testing.T) { require.True(t, statsTbl.Pseudo) require.Zero(t, statsTbl.Version) currentVersion := do.StatsHandle().MaxTableStatsVersion() - testKit.MustExec("analyze table t") + tk.MustExec("analyze table t") statsTbl = do.StatsHandle().GetPhysicalTableStats(tableInfo.ID, tableInfo) require.False(t, statsTbl.Pseudo) require.NotZero(t, statsTbl.Version) @@ -193,11 +195,37 @@ func TestStatsCacheProcess(t *testing.T) { require.Equal(t, currentVersion, newVersion, "analyze should not move forward the stats cache version") // Insert more rows - testKit.MustExec("insert into t values(2, 3)") - testKit.MustExec("flush stats_delta *.*") + tk.MustExec("insert into t values(2, 3)") + tk.MustExec("flush stats_delta *.*") require.NoError(t, do.StatsHandle().Update(context.Background(), is)) newVersion = do.StatsHandle().MaxTableStatsVersion() require.NotEqual(t, currentVersion, newVersion, "update with no table should move forward the stats cache version") + + // Reproduce a targeted ANALYZE refresh after an unrelated table has advanced + // the cache-wide version: cache pseudo stats, advance the max with another + // table, then require the requested lower-version row to be loaded. + const ( + analyzedTableVersion uint64 = 100 + cacheWideMaxVersion uint64 = 200 + ) + tk.MustExec("update mysql.stats_meta set version = ? where table_id = ?", analyzedTableVersion, tableInfo.ID) + handle := do.StatsHandle() + handle.Clear() + statsTbl = handle.GetPhysicalTableStats(tableInfo.ID, tableInfo) + require.True(t, statsTbl.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()) + + require.NoError(t, handle.Update(context.Background(), is, tableInfo.ID)) + statsTbl = handle.GetPhysicalTableStats(tableInfo.ID, tableInfo) + require.False(t, statsTbl.Pseudo) + require.Equal(t, analyzedTableVersion, statsTbl.Version) + require.Equal(t, cacheWideMaxVersion, handle.MaxTableStatsVersion()) } func TestStatsCache(t *testing.T) { diff --git a/pkg/statistics/handle/types/interfaces.go b/pkg/statistics/handle/types/interfaces.go index 6ad61fdcb4c10..77187d3602a23 100644 --- a/pkg/statistics/handle/types/interfaces.go +++ b/pkg/statistics/handle/types/interfaces.go @@ -470,11 +470,12 @@ type GlobalStatsInfo struct { // StatsGlobal is used to manage partition table global stats. type StatsGlobal interface { // MergePartitionStats2GlobalStatsByTableID merges partition stats to global stats by table ID. + // The returned boolean reports whether at least one global stats record was persisted. MergePartitionStats2GlobalStatsByTableID(sc sessionctx.Context, opts map[ast.AnalyzeOptionType]uint64, is infoschema.InfoSchema, info *GlobalStatsInfo, physicalID int64, - ) (err error) + ) (updated bool, err error) } // DDL is used to handle ddl events. From bf3197d1aec473b3e477dc3e525377b8c96e7b65 Mon Sep 17 00:00:00 2001 From: wlwilliamx Date: Wed, 22 Jul 2026 15:38:28 +0800 Subject: [PATCH 2/2] statistics: refresh logical table meta after analyze --- pkg/executor/analyze.go | 24 +++++- pkg/executor/test/analyzetest/analyze_test.go | 4 +- pkg/statistics/handle/cache/statscache.go | 82 ++++++++++++------- pkg/statistics/handle/types/interfaces.go | 4 + 4 files changed, 81 insertions(+), 33 deletions(-) diff --git a/pkg/executor/analyze.go b/pkg/executor/analyze.go index 8627a73300a0b..5b5a02df72e0c 100644 --- a/pkg/executor/analyze.go +++ b/pkg/executor/analyze.go @@ -324,12 +324,17 @@ func (e *AnalyzeExec) Next(ctx context.Context, _ *chunk.Chunk) (err error) { if len(tasks) == 0 { return nil } - // Analyze results are persisted under physical statistics IDs. The logical - // table ID is added later only when global stats are actually persisted. + // 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) statsIDsToRefresh = append(statsIDsToRefresh, tableID.GetStatisticsID()) + if tableID.IsPartitionTable() { + logicalTableMetaIDsToRefresh[tableID.TableID] = struct{}{} + } } // Get the min number of goroutines for parallel execution. @@ -422,7 +427,11 @@ TASKLOOP: }) // If we enabled dynamic prune mode, then we need to generate global stats here for partition tables. if needGlobalStats { - statsIDsToRefresh = append(statsIDsToRefresh, e.handleGlobalStats(statsHandle, globalStatsMap)...) + globalStatsIDs := e.handleGlobalStats(statsHandle, globalStatsMap) + statsIDsToRefresh = append(statsIDsToRefresh, globalStatsIDs...) + for _, tableID := range globalStatsIDs { + delete(logicalTableMetaIDsToRefresh, tableID) + } } if intest.EnableInternalCheck { @@ -442,7 +451,14 @@ TASKLOOP: if err != nil { sessionVars.StmtCtx.AppendWarning(err) } - return statsHandle.Update(ctx, infoSchema, statsIDsToRefresh...) + 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 { diff --git a/pkg/executor/test/analyzetest/analyze_test.go b/pkg/executor/test/analyzetest/analyze_test.go index b4f9293f3ef71..3f68b097e01dc 100644 --- a/pkg/executor/test/analyzetest/analyze_test.go +++ b/pkg/executor/test/analyzetest/analyze_test.go @@ -132,7 +132,8 @@ 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 the global entry against the merge result. +// 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 @@ -230,6 +231,7 @@ func TestAnalyzeRefreshesOnlyPersistedStats(t *testing.T) { } else { require.True(t, globalStats.Pseudo) } + require.Equal(t, int64(4), globalStats.RealtimeCount) require.Equal(t, cacheWideMaxVersion, handle.MaxTableStatsVersion()) }) } diff --git a/pkg/statistics/handle/cache/statscache.go b/pkg/statistics/handle/cache/statscache.go index 0c58782f6821a..71e20b7417cbd 100644 --- a/pkg/statistics/handle/cache/statscache.go +++ b/pkg/statistics/handle/cache/statscache.go @@ -120,6 +120,18 @@ func newCacheOfBatchUpdate(batchSize int, op func(toUpdate []*statistics.Table, // Update reads stats meta from store and updates the stats map. func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, tableAndPartitionIDs ...int64) error { + return s.update(ctx, is, false, tableAndPartitionIDs...) +} + +// UpdateTableStatsMeta refreshes only table metadata while preserving cached histograms and pseudo state. +func (s *StatsCacheImpl) UpdateTableStatsMeta(ctx context.Context, is infoschema.InfoSchema, tableIDs ...int64) error { + if len(tableIDs) == 0 { + return nil + } + return s.update(ctx, is, true, tableIDs...) +} + +func (s *StatsCacheImpl) update(ctx context.Context, is infoschema.InfoSchema, metaOnly bool, tableAndPartitionIDs ...int64) error { onlyForAnalyzedTables := len(tableAndPartitionIDs) > 0 start := time.Now() defer func() { @@ -208,37 +220,51 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t continue } var tbl *statistics.Table - needLoadColAndIdxStats := true - // If the column/index stats has not been updated, we can reuse the old table stats. - // Only need to update the count and modify count. - if ok && latestHistUpdateVersion > 0 && oldTbl.LastStatsHistVersion >= latestHistUpdateVersion { - tbl = oldTbl.CopyAs(statistics.MetaOnly) - // count and modify count is updated in finalProcess - needLoadColAndIdxStats = false - } - if needLoadColAndIdxStats { - tbl, err = s.statsHandle.TableStatsFromStorage( - tableInfo, - physicalID, - false, - 0, - ) - // Error is not nil may mean that there are some ddl changes on this table, we will not update it. - if err != nil { - statslogutil.StatsLogger().Warn( - "error occurred when read table stats", - zap.String("table", tableInfo.Name.O), - zap.Error(err), - ) - continue + if metaOnly { + // A partition ANALYZE always flushes the logical table's row count, + // but it may not persist a new global histogram. Preserve the cached + // histogram and pseudo state instead of loading an older global row. + if ok { + tbl = oldTbl.CopyAs(statistics.MetaOnly) + } else { + tbl = statistics.PseudoTable(tableInfo, false, true) + tbl.PhysicalID = physicalID + } + } else { + needLoadColAndIdxStats := true + // If the column/index stats has not been updated, we can reuse the old table stats. + // Only need to update the count and modify count. + if ok && latestHistUpdateVersion > 0 && oldTbl.LastStatsHistVersion >= latestHistUpdateVersion { + tbl = oldTbl.CopyAs(statistics.MetaOnly) + // count and modify count is updated in finalProcess + needLoadColAndIdxStats = false } - if tbl == nil { - tblToUpdateOrDelete.addToDelete(physicalID) - continue + if needLoadColAndIdxStats { + tbl, err = s.statsHandle.TableStatsFromStorage( + tableInfo, + physicalID, + false, + 0, + ) + // Error is not nil may mean that there are some ddl changes on this table, we will not update it. + if err != nil { + statslogutil.StatsLogger().Warn( + "error occurred when read table stats", + zap.String("table", tableInfo.Name.O), + zap.Error(err), + ) + continue + } + if tbl == nil { + tblToUpdateOrDelete.addToDelete(physicalID) + continue + } } } tbl.Version = version - tbl.LastStatsHistVersion = latestHistUpdateVersion + if !metaOnly { + tbl.LastStatsHistVersion = latestHistUpdateVersion + } tbl.RealtimeCount = count tbl.ModifyCount = modifyCount tbl.TblInfoUpdateTS = tableInfo.UpdateTS @@ -249,7 +275,7 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t // 2. LastAnalyzeVersion is 0 because it has never been loaded. // In this case, we can initialize LastAnalyzeVersion to the snapshot, // otherwise auto-analyze will assume that the table has never been analyzed and try to analyze it again. - if tbl.LastAnalyzeVersion == 0 && snapshot != 0 { + if !metaOnly && tbl.LastAnalyzeVersion == 0 && snapshot != 0 { tbl.LastAnalyzeVersion = snapshot } tblToUpdateOrDelete.addToUpdate(tbl) diff --git a/pkg/statistics/handle/types/interfaces.go b/pkg/statistics/handle/types/interfaces.go index 77187d3602a23..e12d7ab8eac87 100644 --- a/pkg/statistics/handle/types/interfaces.go +++ b/pkg/statistics/handle/types/interfaces.go @@ -226,6 +226,10 @@ type StatsCache interface { // To work with auto-analyze's needs, we'll update all table's stats meta into memory. Update(ctx context.Context, is infoschema.InfoSchema, tableAndPartitionIDs ...int64) error + // UpdateTableStatsMeta refreshes only the count and modify count for the requested tables. + // It preserves cached histograms and the pseudo state when ANALYZE did not persist new global stats. + UpdateTableStatsMeta(ctx context.Context, is infoschema.InfoSchema, tableIDs ...int64) error + // MemConsumed returns its memory usage. MemConsumed() (size int64)