plan, tidb: support plan cache for SELECT statement

config/config.go

@@ -36,6 +36,7 @@ type Config struct {
 	Status      Status      `toml:"status" json:"status"`
 	Performance Performance `toml:"performance" json:"performance"`
 	XProtocol   XProtocol   `toml:"xprotocol" json:"xprotocol"`
+	PlanCache   PlanCache   `toml:"plan-cache" json:"plan-cache"`
 }
 
 // Log is the log section of config.
@@ -88,6 +89,13 @@ type XProtocol struct {
 	XSocket string `toml:"xsocket" json:"xsocket"`
 }
 
+// PlanCache is the PlanCache section of the config.
+type PlanCache struct {
+	Enabled  bool  `toml:"plan-cache-enabled" json:"plan-cache-enabled"`
+	Capacity int64 `toml:"plan-cache-capacity" json:"plan-cache-capacity"`
+	Shards   int64 `toml:"plan-cache-shards" json:"plan-cache-shards"`
+}
+
 var defaultConf = Config{
 	Host:   "0.0.0.0",
 	Port:   4000,
@@ -121,6 +129,11 @@ var defaultConf = Config{
 		XHost: "0.0.0.0",
 		XPort: 14000,
 	},
+	PlanCache: PlanCache{
+		Enabled:  true,
+		Capacity: 2560,
+		Shards:   256,
+	},
 }
 
 var globalConf = defaultConf

config/config.toml.example

@@ -114,3 +114,8 @@ xport = 14000
 
 # The socket file to use for x protocol connection.
 xsocket = ""
+
+[plan-cache]
+plan-cache-enabled = true
+plan-cache-capacity = 2560
+plan-cache-shards = 256

ddl/ddl_db_change_test.go

@@ -242,15 +242,15 @@ func (t *testExecInfo) parseSQLs(p *parser.Parser) error {
 	return nil
 }
 
-func (t *testExecInfo) compileSQL(idx int) error {
-	var err error
+func (t *testExecInfo) compileSQL(idx int) (err error) {
 	compiler := executor.Compiler{}
 	for _, info := range t.sqlInfos {
 		c := info.cases[idx]
 		se := c.session
 		se.PrepareTxnCtx()
 		ctx := se.(context.Context)
 		executor.ResetStmtCtx(ctx, c.rawStmt)
+
 		c.stmt, err = compiler.Compile(ctx, c.rawStmt)
 		if err != nil {
 			return errors.Trace(err)

executor/adapter.go

@@ -40,7 +40,7 @@ type processinfoSetter interface {
 type recordSet struct {
 	fields      []*ast.ResultField
 	executor    Executor
-	stmt        *statement
+	stmt        *ExecStmt
 	processinfo processinfoSetter
 	err         error
 }
@@ -90,35 +90,43 @@ func (a *recordSet) Close() error {
 	return errors.Trace(err)
 }
 
-// statement implements the ast.Statement interface, it builds a plan.Plan to an ast.Statement.
-type statement struct {
-	is infoschema.InfoSchema // The InfoSchema cannot change during execution, so we hold a reference to it.
+// ExecStmt implements the ast.Statement interface, it builds a plan.Plan to an ast.Statement.
+type ExecStmt struct {
+	// InfoSchema stores a reference to the schema information.
+	InfoSchema infoschema.InfoSchema
+	// Plan stores a reference to the final physical plan.
+	Plan plan.Plan
+	// Expensive represents whether this query is an expensive one.
+	Expensive bool
+	// Cacheable represents whether the physical plan can be cached.
+	Cacheable bool
+	// Text represents the origin query text.
+	Text string
 
 	ctx            context.Context
-	text           string
-	plan           plan.Plan
 	startTime      time.Time
 	isPreparedStmt bool
-	expensive      bool
 }
 
-func (a *statement) OriginText() string {
-	return a.text
+// OriginText implements ast.Statement interface.
+func (a *ExecStmt) OriginText() string {
+	return a.Text
 }
 
-func (a *statement) IsPrepared() bool {
+// IsPrepared implements ast.Statement interface.
+func (a *ExecStmt) IsPrepared() bool {
 	return a.isPreparedStmt
 }
 
 // Exec implements the ast.Statement Exec interface.
 // This function builds an Executor from a plan. If the Executor doesn't return result,
 // like the INSERT, UPDATE statements, it executes in this function, if the Executor returns
 // result, execution is done after this function returns, in the returned ast.RecordSet Next method.
-func (a *statement) Exec(ctx context.Context) (ast.RecordSet, error) {
+func (a *ExecStmt) Exec(ctx context.Context) (ast.RecordSet, error) {
 	a.startTime = time.Now()
 	a.ctx = ctx
 
-	if _, ok := a.plan.(*plan.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
+	if _, ok := a.Plan.(*plan.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
 		oriStats := ctx.GetSessionVars().Systems[variable.TiDBBuildStatsConcurrency]
 		oriScan := ctx.GetSessionVars().DistSQLScanConcurrency
 		oriIndex := ctx.GetSessionVars().IndexSerialScanConcurrency
@@ -148,7 +156,7 @@ func (a *statement) Exec(ctx context.Context) (ast.RecordSet, error) {
 	if raw, ok := ctx.(processinfoSetter); ok {
 		pi = raw
 		sql := a.OriginText()
-		if simple, ok := a.plan.(*plan.Simple); ok && simple.Statement != nil {
+		if simple, ok := a.Plan.(*plan.Simple); ok && simple.Statement != nil {
 			if ss, ok := simple.Statement.(ast.SensitiveStmtNode); ok {
 				// Use SecureText to avoid leak password information.
 				sql = ss.SecureText()
@@ -169,7 +177,7 @@ func (a *statement) Exec(ctx context.Context) (ast.RecordSet, error) {
 	}, nil
 }
 
-func (a *statement) handleNoDelayExecutor(e Executor, ctx context.Context, pi processinfoSetter) (ast.RecordSet, error) {
+func (a *ExecStmt) handleNoDelayExecutor(e Executor, ctx context.Context, pi processinfoSetter) (ast.RecordSet, error) {
 	// Check if "tidb_snapshot" is set for the write executors.
 	// In history read mode, we can not do write operations.
 	switch e.(type) {
@@ -203,18 +211,18 @@ func (a *statement) handleNoDelayExecutor(e Executor, ctx context.Context, pi pr
 }
 
 // buildExecutor build a executor from plan, prepared statement may need additional procedure.
-func (a *statement) buildExecutor(ctx context.Context) (Executor, error) {
+func (a *ExecStmt) buildExecutor(ctx context.Context) (Executor, error) {
 	priority := kv.PriorityNormal
-	if _, ok := a.plan.(*plan.Execute); !ok {
+	if _, ok := a.Plan.(*plan.Execute); !ok {
 		// Do not sync transaction for Execute statement, because the real optimization work is done in
 		// "ExecuteExec.Build".
 		var err error
-		isPointGet := IsPointGetWithPKOrUniqueKeyByAutoCommit(ctx, a.plan)
+		isPointGet := IsPointGetWithPKOrUniqueKeyByAutoCommit(ctx, a.Plan)
 		if isPointGet {
-			log.Debugf("[%d][InitTxnWithStartTS] %s", ctx.GetSessionVars().ConnectionID, a.text)
+			log.Debugf("[%d][InitTxnWithStartTS] %s", ctx.GetSessionVars().ConnectionID, a.Text)
 			err = ctx.InitTxnWithStartTS(math.MaxUint64)
 		} else {
-			log.Debugf("[%d][ActivePendingTxn] %s", ctx.GetSessionVars().ConnectionID, a.text)
+			log.Debugf("[%d][ActivePendingTxn] %s", ctx.GetSessionVars().ConnectionID, a.Text)
 			err = ctx.ActivePendingTxn()
 		}
 		if err != nil {
@@ -227,17 +235,17 @@ func (a *statement) buildExecutor(ctx context.Context) (Executor, error) {
 			switch {
 			case isPointGet:
 				priority = kv.PriorityHigh
-			case a.expensive:
+			case a.Expensive:
 				priority = kv.PriorityLow
 			}
 		}
 	}
-	if _, ok := a.plan.(*plan.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
+	if _, ok := a.Plan.(*plan.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
 		priority = kv.PriorityLow
 	}
 
-	b := newExecutorBuilder(ctx, a.is, priority)
-	e := b.build(a.plan)
+	b := newExecutorBuilder(ctx, a.InfoSchema, priority)
+	e := b.build(a.Plan)
 	if b.err != nil {
 		return nil, errors.Trace(b.err)
 	}
@@ -248,18 +256,18 @@ func (a *statement) buildExecutor(ctx context.Context) (Executor, error) {
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
-		a.text = executorExec.Stmt.Text()
+		a.Text = executorExec.Stmt.Text()
 		a.isPreparedStmt = true
-		a.plan = executorExec.Plan
+		a.Plan = executorExec.Plan
 		e = executorExec.StmtExec
 	}
 	return e, nil
 }
 
-func (a *statement) logSlowQuery() {
+func (a *ExecStmt) logSlowQuery() {
 	cfg := config.GetGlobalConfig()
 	costTime := time.Since(a.startTime)
-	sql := a.text
+	sql := a.Text
 	if len(sql) > cfg.Log.QueryLogMaxLen {
 		sql = sql[:cfg.Log.QueryLogMaxLen] + fmt.Sprintf("(len:%d)", len(sql))
 	}

executor/builder.go

@@ -1207,7 +1207,7 @@ func (b *executorBuilder) buildIndexLookUpReader(v *plan.PhysicalIndexLookUpRead
 		tableReaderSchema *expression.Schema
 	)
 	table, _ := b.is.TableByID(is.Table.ID)
-	len := v.Schema().Len()
+	length := v.Schema().Len()
 	if v.NeedColHandle {
 		handleCol = v.Schema().TblID2Handle[is.Table.ID][0]
 	} else if !is.OutOfOrder {
@@ -1219,13 +1219,18 @@ func (b *executorBuilder) buildIndexLookUpReader(v *plan.PhysicalIndexLookUpRead
 			RetType: types.NewFieldType(mysql.TypeLonglong),
 		}
 		tableReaderSchema.Append(handleCol)
-		len = tableReaderSchema.Len()
+		length = tableReaderSchema.Len()
 	}
 
-	for i := 0; i < len; i++ {
+	for i := 0; i < length; i++ {
 		tableReq.OutputOffsets = append(tableReq.OutputOffsets, uint32(i))
 	}
 
+	ranges := make([]*types.IndexRange, 0, len(is.Ranges))
+	for _, rangeInPlan := range is.Ranges {
+		ranges = append(ranges, rangeInPlan.Clone())
+	}
+
 	e := &IndexLookUpExecutor{
 		ctx:               b.ctx,
 		schema:            v.Schema(),
@@ -1235,7 +1240,7 @@ func (b *executorBuilder) buildIndexLookUpReader(v *plan.PhysicalIndexLookUpRead
 		index:             is.Index,
 		keepOrder:         !is.OutOfOrder,
 		desc:              is.Desc,
-		ranges:            is.Ranges,
+		ranges:            ranges,
 		tableRequest:      tableReq,
 		columns:           is.Columns,
 		handleCol:         handleCol,

executor/compiler.go

@@ -22,36 +22,33 @@ import (
 	"github.com/pingcap/tidb/plan"
 )
 
-// Compiler compiles an ast.StmtNode to a stmt.Statement.
+// Compiler compiles an ast.StmtNode to a physical plan.
 type Compiler struct {
 }
 
-// Compile compiles an ast.StmtNode to an ast.Statement.
-// After preprocessed and validated, it will be optimized to a plan,
-// then wrappped to an adapter *statement as stmt.Statement.
-func (c *Compiler) Compile(ctx context.Context, node ast.StmtNode) (ast.Statement, error) {
-	is := GetInfoSchema(ctx)
-	if err := plan.Preprocess(node, is, ctx); err != nil {
+// Compile compiles an ast.StmtNode to a physical plan.
+func (c *Compiler) Compile(ctx context.Context, stmtNode ast.StmtNode) (*ExecStmt, error) {
+	infoSchema := GetInfoSchema(ctx)
+	if err := plan.Preprocess(stmtNode, infoSchema, ctx); err != nil {
 		return nil, errors.Trace(err)
 	}
 	// Validate should be after NameResolve.
-	if err := plan.Validate(node, false); err != nil {
+	if err := plan.Validate(stmtNode, false); err != nil {
 		return nil, errors.Trace(err)
 	}
-	p, err := plan.Optimize(ctx, node, is)
+
+	finalPlan, err := plan.Optimize(ctx, stmtNode, infoSchema)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	// Don't take restricted SQL into account for metrics.
-	isExpensive := stmtCount(node, p, ctx.GetSessionVars().InRestrictedSQL)
-	sa := &statement{
-		is:        is,
-		plan:      p,
-		text:      node.Text(),
-		expensive: isExpensive,
-	}
-	return sa, nil
+	return &ExecStmt{
+		InfoSchema: infoSchema,
+		Plan:       finalPlan,
+		Expensive:  stmtCount(stmtNode, finalPlan, ctx.GetSessionVars().InRestrictedSQL),
+		Cacheable:  plan.Cacheable(stmtNode),
+		Text:       stmtNode.Text(),
+	}, nil
 }
 
 // GetInfoSchema gets TxnCtx InfoSchema if snapshot schema is not set,

executor/prepared.go

@@ -318,14 +318,14 @@ func CompileExecutePreparedStmt(ctx context.Context, ID uint32, args ...interfac
 		value := ast.NewValueExpr(val)
 		execPlan.UsingVars[i] = &expression.Constant{Value: value.Datum, RetType: &value.Type}
 	}
-	sa := &statement{
-		is:   GetInfoSchema(ctx),
-		plan: execPlan,
+	stmt := &ExecStmt{
+		InfoSchema: GetInfoSchema(ctx),
+		Plan:       execPlan,
 	}
 	if prepared, ok := ctx.GetSessionVars().PreparedStmts[ID].(*Prepared); ok {
-		sa.text = prepared.Stmt.Text()
+		stmt.Text = prepared.Stmt.Text()
 	}
-	return sa
+	return stmt
 }
 
 // ResetStmtCtx resets the StmtContext.