executor: support fast analyze.

executor/builder.go

@@ -1430,20 +1430,49 @@ func (b *executorBuilder) buildAnalyze(v *plannercore.Analyze) Executor {
 		tasks:        make([]*analyzeTask, 0, len(v.ColTasks)+len(v.IdxTasks)),
 	}
 	for _, task := range v.ColTasks {
-		e.tasks = append(e.tasks, &analyzeTask{
-			taskType: colTask,
-			colExec:  b.buildAnalyzeColumnsPushdown(task, v.MaxNumBuckets),
-		})
+		if v.EnableFastAnalyze {
+			e.tasks = append(e.tasks, &analyzeTask{
+				taskType: fastTask,
+				fastExec: &AnalyzeFastExec{
+					ctx:             b.ctx,
+					PhysicalTableID: task.PhysicalTableID,
+					colsInfo:        task.ColsInfo,
+					pkInfo:          task.PKInfo,
+					maxNumBuckets:   v.MaxNumBuckets,
+					table:           task.Table,
+					concurrency:     4,
+				},
+			})
+		} else {
+			e.tasks = append(e.tasks, &analyzeTask{
+				taskType: colTask,
+				colExec:  b.buildAnalyzeColumnsPushdown(task, v.MaxNumBuckets),
+			})
+		}
 		if b.err != nil {
 			b.err = errors.Trace(b.err)
 			return nil
 		}
 	}
 	for _, task := range v.IdxTasks {
-		e.tasks = append(e.tasks, &analyzeTask{
-			taskType: idxTask,
-			idxExec:  b.buildAnalyzeIndexPushdown(task, v.MaxNumBuckets),
-		})
+		if v.EnableFastAnalyze {
+			e.tasks = append(e.tasks, &analyzeTask{
+				taskType: fastTask,
+				fastExec: &AnalyzeFastExec{
+					ctx:             b.ctx,
+					PhysicalTableID: task.PhysicalTableID,
+					idxInfo:         task.IndexInfo,
+					maxNumBuckets:   v.MaxNumBuckets,
+					table:           task.Table,
+					concurrency:     4,
+				},
+			})
+		} else {
+			e.tasks = append(e.tasks, &analyzeTask{
+				taskType: idxTask,
+				idxExec:  b.buildAnalyzeIndexPushdown(task, v.MaxNumBuckets),
+			})
+		}
 		if b.err != nil {
 			b.err = errors.Trace(b.err)
 			return nil

planner/core/common_plans.go

@@ -402,13 +402,15 @@ type AnalyzeColumnsTask struct {
 	PhysicalTableID int64
 	PKInfo          *model.ColumnInfo
 	ColsInfo        []*model.ColumnInfo
+	Table           table.Table
 }
 
 // AnalyzeIndexTask is used for analyze index.
 type AnalyzeIndexTask struct {
 	// PhysicalTableID is the id for a partition or a table.
 	PhysicalTableID int64
 	IndexInfo       *model.IndexInfo
+	Table           table.Table
 }
 
 // Analyze represents an analyze plan
@@ -418,6 +420,9 @@ type Analyze struct {
 	ColTasks      []AnalyzeColumnsTask
 	IdxTasks      []AnalyzeIndexTask
 	MaxNumBuckets uint64
+	TableNames    []*ast.TableName
+
+	EnableFastAnalyze bool
 }
 
 // LoadData represents a loaddata plan.

planner/core/planbuilder.go

@@ -742,7 +742,10 @@ func getPhysicalIDs(tblInfo *model.TableInfo, partitionNames []model.CIStr) ([]i
 }
 
 func (b *PlanBuilder) buildAnalyzeTable(as *ast.AnalyzeTableStmt) (Plan, error) {
-	p := &Analyze{MaxNumBuckets: as.MaxNumBuckets}
+	p := &Analyze{
+		MaxNumBuckets:     as.MaxNumBuckets,
+		EnableFastAnalyze: b.ctx.GetSessionVars().EnableFastAnalyze,
+	}
 	for _, tbl := range as.TableNames {
 		if tbl.TableInfo.IsView() {
 			return nil, errors.Errorf("analyze %s is not supported now.", tbl.Name.O)
@@ -752,22 +755,38 @@ func (b *PlanBuilder) buildAnalyzeTable(as *ast.AnalyzeTableStmt) (Plan, error)
 		if err != nil {
 			return nil, err
 		}
+		table, ok := b.is.TableByID(tbl.TableInfo.ID)
+		if !ok {
+			return nil, infoschema.ErrTableNotExists.GenWithStackByArgs(tbl.DBInfo.Name.O, tbl.TableInfo.Name.O)
+		}
 		for _, idx := range idxInfo {
 			for _, id := range physicalIDs {
-				p.IdxTasks = append(p.IdxTasks, AnalyzeIndexTask{PhysicalTableID: id, IndexInfo: idx})
+				p.IdxTasks = append(p.IdxTasks, AnalyzeIndexTask{
+					PhysicalTableID: id,
+					IndexInfo:       idx,
+					Table:           table,
+				})
 			}
 		}
 		if len(colInfo) > 0 || pkInfo != nil {
 			for _, id := range physicalIDs {
-				p.ColTasks = append(p.ColTasks, AnalyzeColumnsTask{PhysicalTableID: id, PKInfo: pkInfo, ColsInfo: colInfo})
+				p.ColTasks = append(p.ColTasks, AnalyzeColumnsTask{
+					PhysicalTableID: id,
+					PKInfo:          pkInfo,
+					ColsInfo:        colInfo,
+					Table:           table,
+				})
 			}
 		}
 	}
 	return p, nil
 }
 
 func (b *PlanBuilder) buildAnalyzeIndex(as *ast.AnalyzeTableStmt) (Plan, error) {
-	p := &Analyze{MaxNumBuckets: as.MaxNumBuckets}
+	p := &Analyze{
+		MaxNumBuckets:     as.MaxNumBuckets,
+		EnableFastAnalyze: b.ctx.GetSessionVars().EnableFastAnalyze,
+	}
 	tblInfo := as.TableNames[0].TableInfo
 	physicalIDs, err := getPhysicalIDs(tblInfo, as.PartitionNames)
 	if err != nil {
@@ -786,7 +805,10 @@ func (b *PlanBuilder) buildAnalyzeIndex(as *ast.AnalyzeTableStmt) (Plan, error)
 }
 
 func (b *PlanBuilder) buildAnalyzeAllIndex(as *ast.AnalyzeTableStmt) (Plan, error) {
-	p := &Analyze{MaxNumBuckets: as.MaxNumBuckets}
+	p := &Analyze{
+		MaxNumBuckets:     as.MaxNumBuckets,
+		EnableFastAnalyze: b.ctx.GetSessionVars().EnableFastAnalyze,
+	}
 	tblInfo := as.TableNames[0].TableInfo
 	physicalIDs, err := getPhysicalIDs(tblInfo, as.PartitionNames)
 	if err != nil {

sessionctx/variable/session.go

@@ -340,6 +340,9 @@ type SessionVars struct {
 
 	// SlowQueryFile indicates which slow query log file for SLOW_QUERY table to parse.
 	SlowQueryFile string
+
+	// EnableFastAnalyze indicates whether to take fast analyze.
+	EnableFastAnalyze bool
 }
 
 // ConnectionInfo present connection used by audit.
@@ -387,6 +390,7 @@ func NewSessionVars() *SessionVars {
 		L2CacheSize:               cpuid.CPU.Cache.L2,
 		CommandValue:              uint32(mysql.ComSleep),
 		SlowQueryFile:             config.GetGlobalConfig().Log.SlowQueryFile,
+		EnableFastAnalyze:         false,
 	}
 	vars.Concurrency = Concurrency{
 		IndexLookupConcurrency:     DefIndexLookupConcurrency,
@@ -725,6 +729,8 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		config.GetGlobalConfig().CheckMb4ValueInUTF8 = TiDBOptOn(val)
 	case TiDBSlowQueryFile:
 		s.SlowQueryFile = val
+	case TiDBEnableFastAnalyze:
+		s.EnableFastAnalyze = TiDBOptOn(val)
 	}
 	s.systems[name] = val
 	return nil

sessionctx/variable/sysvar.go

@@ -686,6 +686,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TiDBEnableRadixJoin, BoolToIntStr(DefTiDBUseRadixJoin)},
 	{ScopeSession, TiDBCheckMb4ValueInUTF8, BoolToIntStr(config.GetGlobalConfig().CheckMb4ValueInUTF8)},
 	{ScopeSession, TiDBSlowQueryFile, ""},
+	{ScopeSession, TiDBEnableFastAnalyze, BoolToIntStr(DefTiDBUseFastAnalyze)},
 }
 
 // SynonymsSysVariables is synonyms of system variables.

sessionctx/variable/tidb_vars.go

@@ -239,6 +239,9 @@ const (
 
 	// SlowQueryFile indicates which slow query log file for SLOW_QUERY table to parse.
 	TiDBSlowQueryFile = "tidb_slow_query_file"
+
+	// TiDBEnableFastAnalyze indicates to use fast analyze.
+	TiDBEnableFastAnalyze = "tidb_enable_fast_analyze"
 )
 
 // Default TiDB system variable values.
@@ -291,6 +294,7 @@ const (
 	DefTiDBUseRadixJoin              = false
 	DefEnableWindowFunction          = false
 	DefTiDBDDLSlowOprThreshold       = 300
+	DefTiDBUseFastAnalyze            = false
 )
 
 // Process global variables.

sessionctx/variable/varsutil_test.go

@@ -79,6 +79,7 @@ func (s *testVarsutilSuite) TestNewSessionVars(c *C) {
 	c.Assert(vars.MemQuotaNestedLoopApply, Equals, int64(DefTiDBMemQuotaNestedLoopApply))
 	c.Assert(vars.EnableRadixJoin, Equals, DefTiDBUseRadixJoin)
 	c.Assert(vars.AllowWriteRowID, Equals, DefOptWriteRowID)
+	c.Assert(vars.EnableFastAnalyze, Equals, DefTiDBUseFastAnalyze)
 
 	assertFieldsGreaterThanZero(c, reflect.ValueOf(vars.Concurrency))
 	assertFieldsGreaterThanZero(c, reflect.ValueOf(vars.MemQuota))

store/tikv/client.go

@@ -22,6 +22,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/debugpb"
+
 	"github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
 	"github.com/grpc-ecosystem/go-grpc-prometheus"
@@ -579,6 +581,17 @@ func (c *rpcClient) SendRequest(ctx context.Context, addr string, req *tikvrpc.R
 		}
 	}
 
+	if req.IsDebugReq() {
+		client := debugpb.NewDebugClient(connArray.Get())
+		ctx1, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
+		resp, err := tikvrpc.CallDebugRPC(ctx1, client, req)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		return resp, nil
+	}
+
 	client := tikvpb.NewTikvClient(connArray.Get())
 
 	if req.Type != tikvrpc.CmdCopStream {

store/tikv/region_cache.go

@@ -219,7 +219,7 @@ func (c *RegionCache) LocateRegionByID(bo *Backoffer, regionID uint64) (*KeyLoca
 	}
 	c.mu.RUnlock()
 
-	r, err := c.loadRegionByID(bo, regionID)
+	r, err := c.LoadRegionByID(bo, regionID)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -420,8 +420,8 @@ func (c *RegionCache) loadRegion(bo *Backoffer, key []byte, isEndKey bool) (*Reg
 	}
 }
 
-// loadRegionByID loads region from pd client, and picks the first peer as leader.
-func (c *RegionCache) loadRegionByID(bo *Backoffer, regionID uint64) (*Region, error) {
+// LoadRegionByID loads region from pd client, and picks the first peer as leader.
+func (c *RegionCache) LoadRegionByID(bo *Backoffer, regionID uint64) (*Region, error) {
 	var backoffErr error
 	for {
 		if backoffErr != nil {