@@ -34,6 +34,7 @@ import (
3434 "github.com/pingcap/tidb/pkg/kv"
3535 "github.com/pingcap/tidb/pkg/metrics"
3636 "github.com/pingcap/tidb/pkg/parser/ast"
37+ pmodel "github.com/pingcap/tidb/pkg/parser/model"
3738 "github.com/pingcap/tidb/pkg/planner/core"
3839 "github.com/pingcap/tidb/pkg/sessionctx"
3940 "github.com/pingcap/tidb/pkg/sessionctx/variable"
@@ -84,6 +85,82 @@ const (
8485 idxTask
8586)
8687
88+ // flushStatsDeltaForAnalyze flushes pending stats deltas for the tables whose column-analyze
89+ // tasks will capture base count / modify_count from mysql.stats_meta. Without this, a stale
90+ // pre-analyze delta can be applied later and double count rows or modifications.
91+ func flushStatsDeltaForAnalyze (ctx context.Context , sctx sessionctx.Context , plan * core.Analyze ) error {
92+ if len (plan .ColTasks ) == 0 {
93+ return nil
94+ }
95+ if err := ctx .Err (); err != nil {
96+ return err
97+ }
98+ targetIDs := collectAnalyzeStatsDeltaTargetIDsForTest (plan )
99+ if len (targetIDs ) == 0 {
100+ return nil
101+ }
102+ return domain .GetDomain (sctx ).StatsHandle ().DumpStatsDeltaToKV (true , targetIDs ... )
103+ }
104+
105+ // collectStatsDeltaFlushObjectsForAnalyze returns the database-qualified table
106+ // objects whose stats deltas must be flushed before building column analyze
107+ // tasks. Column analyze captures base count / modify_count from mysql.stats_meta,
108+ // so each target table is included once even if it has multiple column tasks.
109+ func collectStatsDeltaFlushObjectsForAnalyze (plan * core.Analyze ) []* ast.StatsObject {
110+ flushObjects := make ([]* ast.StatsObject , 0 , len (plan .ColTasks ))
111+ type statsObjectKey struct {
112+ dbName string
113+ tableName string
114+ }
115+ seenObjects := make (map [statsObjectKey ]struct {}, len (plan .ColTasks ))
116+ appendFlushObject := func (task core.AnalyzeColumnsTask ) {
117+ dbName , tableName := task .DBName , task .TableName
118+ if dbName == "" || tableName == "" {
119+ intest .Assert (false , "analyze column task must have database-qualified table name" )
120+ return
121+ }
122+ key := statsObjectKey {dbName : dbName , tableName : tableName }
123+ if _ , ok := seenObjects [key ]; ok {
124+ return
125+ }
126+ seenObjects [key ] = struct {}{}
127+ flushObjects = append (flushObjects , & ast.StatsObject {
128+ StatsObjectScope : ast .StatsObjectScopeTable ,
129+ DBName : pmodel .NewCIStr (dbName ),
130+ TableName : pmodel .NewCIStr (tableName ),
131+ })
132+ }
133+ for _ , task := range plan .ColTasks {
134+ appendFlushObject (task )
135+ }
136+ return flushObjects
137+ }
138+
139+ func collectAnalyzeStatsDeltaTargetIDsForTest (plan * core.Analyze ) []int64 {
140+ targetIDs := make ([]int64 , 0 , len (plan .ColTasks ))
141+ seenTargetIDs := make (map [int64 ]struct {}, len (plan .ColTasks ))
142+ appendTargetID := func (id int64 ) {
143+ if _ , ok := seenTargetIDs [id ]; ok {
144+ return
145+ }
146+ seenTargetIDs [id ] = struct {}{}
147+ targetIDs = append (targetIDs , id )
148+ }
149+ for _ , task := range plan .ColTasks {
150+ if task .TblInfo == nil {
151+ intest .Assert (false , "analyze column task must have table info" )
152+ continue
153+ }
154+ appendTargetID (task .TblInfo .ID )
155+ if partitionInfo := task .TblInfo .GetPartitionInfo (); partitionInfo != nil {
156+ for _ , def := range partitionInfo .Definitions {
157+ appendTargetID (def .ID )
158+ }
159+ }
160+ }
161+ return targetIDs
162+ }
163+
87164// Next implements the Executor Next interface.
88165// It will collect all the sample task and run them concurrently.
89166func (e * AnalyzeExec ) Next (ctx context.Context , _ * chunk.Chunk ) error {
0 commit comments