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..5b5a02df72e0c 100644 --- a/pkg/executor/analyze.go +++ b/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 { 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..3f68b097e01dc 100644 --- a/pkg/executor/test/analyzetest/analyze_test.go +++ b/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) diff --git a/pkg/statistics/handle/cache/statscache.go b/pkg/statistics/handle/cache/statscache.go index fd98ba1eb884e..71e20b7417cbd 100644 --- a/pkg/statistics/handle/cache/statscache.go +++ b/pkg/statistics/handle/cache/statscache.go @@ -120,21 +120,32 @@ 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() { 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 +159,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...) @@ -203,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 } - if tbl == nil { - tblToUpdateOrDelete.addToDelete(physicalID) - continue + } 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 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 @@ -244,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/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..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) @@ -470,11 +474,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.