From eba19fe3c7bf2720abb7db0ed6784e6ffa577535 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 27 Dec 2022 16:47:22 +0800
Subject: [PATCH 01/31] commit

---
 executor/prepared.go                   |  2 +-
 planner/core/plan_cache.go             |  4 +-
 planner/core/plan_cache_utils.go       | 56 +++++++++++++++++++++++++-
 planner/core/plan_cacheable_checker.go | 30 +++++++-------
 server/driver_tidb.go                  |  2 +-
 session/session.go                     |  2 +-
 6 files changed, 75 insertions(+), 21 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index 6a5025e0d539b..f9dd9ff541105 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -205,7 +205,7 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL)
+			0, bindSQL, prepared)
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index ab4eb4e4912ab..143c7b2124284 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -145,7 +145,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -285,7 +285,7 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 8dc867316207d..b22103e5a23dc 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -190,6 +190,7 @@ type planCacheKey struct {
 	inRestrictedSQL          bool
 	restrictedReadOnly       bool
 	TiDBSuperReadOnly        bool
+	limitOffsetAndCount      []int64
 
 	memoryUsage int64 // Do not include in hash
 	hash        []byte
@@ -226,6 +227,9 @@ func (key *planCacheKey) Hash() []byte {
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.inRestrictedSQL))...)
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.restrictedReadOnly))...)
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.TiDBSuperReadOnly))...)
+		for _, l := range key.limitOffsetAndCount {
+			key.hash = codec.EncodeInt(key.hash, l)
+		}
 	}
 	return key.hash
 }
@@ -267,7 +271,7 @@ func SetPstmtIDSchemaVersion(key kvcache.Key, stmtText string, schemaVersion int
 // Note: lastUpdatedSchemaVersion will only be set in the case of rc or for update read in order to
 // differentiate the cache key. In other cases, it will be 0.
 func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string, schemaVersion int64,
-	lastUpdatedSchemaVersion int64, bindSQL string) (kvcache.Key, error) {
+	lastUpdatedSchemaVersion int64, bindSQL string, preparedAst *ast.Prepared) (kvcache.Key, error) {
 	if stmtText == "" {
 		return nil, errors.New("no statement text")
 	}
@@ -281,6 +285,7 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 	if sessionVars.TimeZone != nil {
 		_, timezoneOffset = time.Now().In(sessionVars.TimeZone).Zone()
 	}
+	limit := getLimitFromAst(preparedAst.Stmt)
 	key := &planCacheKey{
 		database:                 stmtDB,
 		connID:                   sessionVars.ConnectionID,
@@ -295,6 +300,7 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 		inRestrictedSQL:          sessionVars.InRestrictedSQL,
 		restrictedReadOnly:       variable.RestrictedReadOnly.Load(),
 		TiDBSuperReadOnly:        variable.VarTiDBSuperReadOnly.Load(),
+		limitOffsetAndCount:      limit,
 	}
 	for k, v := range sessionVars.IsolationReadEngines {
 		key.isolationReadEngines[k] = v
@@ -442,3 +448,51 @@ func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCa
 	}
 	return nil, ErrStmtNotFound
 }
+
+type limitExtractor struct {
+	//sctx           sessionctx.Context
+	//schema         infoschema.InfoSchema
+	cacheable      bool
+	hasLimit       bool
+	offsetAndCount []int64
+}
+
+// Enter implements Visitor interface.
+func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bool) {
+	switch node := in.(type) {
+	case *ast.Limit:
+		if node.Count != nil {
+			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
+				//checker.cacheable = false
+				checker.hasLimit = true
+				checker.offsetAndCount = append(checker.offsetAndCount, count.GetInt64())
+				return in, false
+			}
+		}
+		if node.Offset != nil {
+			if offset, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
+				//checker.cacheable = false
+				checker.hasLimit = true
+				checker.offsetAndCount = append(checker.offsetAndCount, offset.GetInt64())
+				return in, false
+			}
+		}
+	}
+	return in, false
+}
+
+// Leave implements Visitor interface.
+func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
+	return in, checker.cacheable
+}
+
+func getLimitFromAst(node ast.Node) []int64 {
+	checker := limitExtractor{
+		//sctx:      sctx,
+		//schema:    is,
+		cacheable:      true,
+		offsetAndCount: make([]int64, 1),
+	}
+	node.Accept(&checker)
+	return checker.offsetAndCount
+}
diff --git a/planner/core/plan_cacheable_checker.go b/planner/core/plan_cacheable_checker.go
index 041509d224792..3eca17c1708e7 100644
--- a/planner/core/plan_cacheable_checker.go
+++ b/planner/core/plan_cacheable_checker.go
@@ -135,21 +135,21 @@ func (checker *cacheableChecker) Enter(in ast.Node) (out ast.Node, skipChildren
 				return in, true
 			}
 		}
-	case *ast.Limit:
-		if node.Count != nil {
-			if _, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-				checker.cacheable = false
-				checker.reason = "query has 'limit ?' is un-cacheable"
-				return in, true
-			}
-		}
-		if node.Offset != nil {
-			if _, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
-				checker.cacheable = false
-				checker.reason = "query has 'limit ?, 10' is un-cacheable"
-				return in, true
-			}
-		}
+	//case *ast.Limit:
+	//	if node.Count != nil {
+	//		if _, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
+	//			checker.cacheable = false
+	//			checker.reason = "query has 'limit ?' is un-cacheable"
+	//			return in, true
+	//		}
+	//	}
+	//	if node.Offset != nil {
+	//		if _, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
+	//			checker.cacheable = false
+	//			checker.reason = "query has 'limit ?, 10' is un-cacheable"
+	//			return in, true
+	//		}
+	//	}
 	case *ast.FrameBound:
 		if _, ok := node.Expr.(*driver.ParamMarkerExpr); ok {
 			checker.cacheable = false
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index 7b25a998d618b..e4e3883060f7f 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -172,7 +172,7 @@ func (ts *TiDBStatement) Close() error {
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL)
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, preparedObj.PreparedAst)
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index d358d761560e2..a4bca0a0b98e9 100644
--- a/session/session.go
+++ b/session/session.go
@@ -383,7 +383,7 @@ func (s *session) cleanRetryInfo() {
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL)
+					0, bindSQL, preparedAst)
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return

From 47c4f5175b5727f5c04328b05b80d2d1d4a7d6de Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 27 Dec 2022 17:07:09 +0800
Subject: [PATCH 02/31] pass stmtNode

---
 executor/prepared.go                  | 2 +-
 planner/core/plan_cache.go            | 4 ++--
 planner/core/plan_cache_utils.go      | 7 +++++--
 planner/core/plan_cache_utils_test.go | 8 ++++----
 server/driver_tidb.go                 | 2 +-
 session/session.go                    | 2 +-
 6 files changed, 14 insertions(+), 11 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index f9dd9ff541105..14b01d32ea3c2 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -205,7 +205,7 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL, prepared)
+			0, bindSQL, prepared.Stmt)
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 143c7b2124284..13167ab4ce3a6 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -145,7 +145,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst.Stmt); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -285,7 +285,7 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst.Stmt); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index b22103e5a23dc..87e9ceb794a41 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -271,7 +271,7 @@ func SetPstmtIDSchemaVersion(key kvcache.Key, stmtText string, schemaVersion int
 // Note: lastUpdatedSchemaVersion will only be set in the case of rc or for update read in order to
 // differentiate the cache key. In other cases, it will be 0.
 func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string, schemaVersion int64,
-	lastUpdatedSchemaVersion int64, bindSQL string, preparedAst *ast.Prepared) (kvcache.Key, error) {
+	lastUpdatedSchemaVersion int64, bindSQL string, stmtNode ast.StmtNode) (kvcache.Key, error) {
 	if stmtText == "" {
 		return nil, errors.New("no statement text")
 	}
@@ -285,7 +285,7 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 	if sessionVars.TimeZone != nil {
 		_, timezoneOffset = time.Now().In(sessionVars.TimeZone).Zone()
 	}
-	limit := getLimitFromAst(preparedAst.Stmt)
+	limit := getLimitFromAst(stmtNode)
 	key := &planCacheKey{
 		database:                 stmtDB,
 		connID:                   sessionVars.ConnectionID,
@@ -487,6 +487,9 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 }
 
 func getLimitFromAst(node ast.Node) []int64 {
+	if node == nil {
+		return []int64{}
+	}
 	checker := limitExtractor{
 		//sctx:      sctx,
 		//schema:    is,
diff --git a/planner/core/plan_cache_utils_test.go b/planner/core/plan_cache_utils_test.go
index 6f0938e447263..2ef6766231312 100644
--- a/planner/core/plan_cache_utils_test.go
+++ b/planner/core/plan_cache_utils_test.go
@@ -32,20 +32,20 @@ func TestCacheKey(t *testing.T) {
 	ctx.GetSessionVars().InRestrictedSQL = false
 	variable.RestrictedReadOnly.Store(false)
 	variable.VarTiDBSuperReadOnly.Store(false)
-	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "")
+	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", nil)
 	if err.Error() != "no statement text" {
 		t.Fail() // no statement text
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "")
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", nil)
 	if err != nil {
 		t.Fail() // schema can be nil
 	}
 	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1,
-		"select /*+ ignore_plan_cache() */ * from t")
+		"select /*+ ignore_plan_cache() */ * from t", nil)
 	if err != nil {
 		t.Fail()
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "")
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", nil)
 	if err != nil {
 		t.Fail()
 	}
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index e4e3883060f7f..ac983edc50585 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -172,7 +172,7 @@ func (ts *TiDBStatement) Close() error {
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, preparedObj.PreparedAst)
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, preparedObj.PreparedAst.Stmt)
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index a4bca0a0b98e9..154337b053211 100644
--- a/session/session.go
+++ b/session/session.go
@@ -383,7 +383,7 @@ func (s *session) cleanRetryInfo() {
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL, preparedAst)
+					0, bindSQL, preparedAst.Stmt)
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return

From 544fb69efeab1730a1d659f12d662a14e16a092e Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 10:10:00 +0800
Subject: [PATCH 03/31] fix ut

---
 planner/core/plan_cacheable_checker_test.go | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/planner/core/plan_cacheable_checker_test.go b/planner/core/plan_cacheable_checker_test.go
index e87a08592eb16..7d417e377888f 100644
--- a/planner/core/plan_cacheable_checker_test.go
+++ b/planner/core/plan_cacheable_checker_test.go
@@ -87,7 +87,7 @@ func TestCacheable(t *testing.T) {
 		TableRefs: tableRefsClause,
 		Limit:     limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{
 		Offset: &driver.ParamMarkerExpr{},
@@ -96,7 +96,7 @@ func TestCacheable(t *testing.T) {
 		TableRefs: tableRefsClause,
 		Limit:     limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{}
 	stmt = &ast.DeleteStmt{
@@ -139,7 +139,7 @@ func TestCacheable(t *testing.T) {
 		TableRefs: tableRefsClause,
 		Limit:     limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{
 		Offset: &driver.ParamMarkerExpr{},
@@ -148,7 +148,7 @@ func TestCacheable(t *testing.T) {
 		TableRefs: tableRefsClause,
 		Limit:     limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{}
 	stmt = &ast.UpdateStmt{
@@ -188,7 +188,7 @@ func TestCacheable(t *testing.T) {
 	stmt = &ast.SelectStmt{
 		Limit: limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{
 		Offset: &driver.ParamMarkerExpr{},
@@ -196,7 +196,7 @@ func TestCacheable(t *testing.T) {
 	stmt = &ast.SelectStmt{
 		Limit: limitStmt,
 	}
-	require.False(t, core.Cacheable(stmt, is))
+	require.True(t, core.Cacheable(stmt, is))
 
 	limitStmt = &ast.Limit{}
 	stmt = &ast.SelectStmt{

From 9ce7ca2e8bd7414741f40147416d02b2fa1a60b2 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 10:18:40 +0800
Subject: [PATCH 04/31] fix ut

---
 planner/core/plan_cache_test.go | 8 +-------
 1 file changed, 1 insertion(+), 7 deletions(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index a480c583d0434..ceb8c211db335 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -282,13 +282,7 @@ func TestPlanCacheDiagInfo(t *testing.T) {
 
 	tk.MustExec("prepare stmt from 'select /*+ ignore_plan_cache() */ * from t'")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: ignore plan cache by hint"))
-
-	tk.MustExec("prepare stmt from 'select * from t limit ?'")
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'limit ?' is un-cacheable"))
-
-	tk.MustExec("prepare stmt from 'select * from t limit ?, 1'")
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'limit ?, 10' is un-cacheable"))
-
+	
 	tk.MustExec("prepare stmt from 'select * from t order by ?'")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'order by ?' is un-cacheable"))
 

From 9a759eee413f424852edcead5d5dc08df5451dd0 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 14:54:19 +0800
Subject: [PATCH 05/31] commit

---
 planner/core/plan_cache_test.go  | 42 +++++++++++++++++++++++++++++++-
 planner/core/plan_cache_utils.go |  8 +++---
 2 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index ceb8c211db335..7479c7a70718e 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -282,7 +282,7 @@ func TestPlanCacheDiagInfo(t *testing.T) {
 
 	tk.MustExec("prepare stmt from 'select /*+ ignore_plan_cache() */ * from t'")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: ignore plan cache by hint"))
-	
+
 	tk.MustExec("prepare stmt from 'select * from t order by ?'")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'order by ?' is un-cacheable"))
 
@@ -296,3 +296,43 @@ func TestPlanCacheDiagInfo(t *testing.T) {
 	tk.MustExec("execute stmt using @a, @b") // a=1 and a=1 -> a=1
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: some parameters may be overwritten"))
 }
+
+func TestPlanCacheWithLimit(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int, b int, key(a))")
+
+	testCases := []struct {
+		sql    string
+		params []int
+	}{
+		{"prepare stmt from 'select * from t limit ?'", []int{1}},
+		{"prepare stmt from 'select * from t limit ?, ?'", []int{1, 2}},
+		{"prepare stmt from 'delete from t order by a limit ?'", []int{1}},
+		{"prepare stmt from 'insert into t select * from t order by a desc limit ?'", []int{1}},
+		{"prepare stmt from 'insert into t select * from t order by a desc limit ?, ?'", []int{1, 2}},
+		{"prepare stmt from 'update t set a = 1 limit ?'", []int{1}},
+		{" prepare stmt from '(select * from t order by a limit ?) union (select * from t order by a desc limit ?)';", []int{1, 2}},
+		{" prepare stmt from 'select * from t where a in (select b from t limit ?) limit ?;';", []int{10, 5}},
+	}
+
+	for _, testCase := range testCases {
+		tk.MustExec(testCase.sql)
+		tk.MustExec("set @a = 1")
+		var using []string
+		for i, p := range testCase.params {
+			tk.MustExec(fmt.Sprintf("set @a%d = %d", i, p))
+			using = append(using, fmt.Sprintf("@a%d", i))
+		}
+
+		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
+		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
+		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("1"))
+
+		tk.MustExec("set @a0 = 10086")
+		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
+		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
+	}
+}
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 87e9ceb794a41..c8dcee84de336 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -465,8 +465,9 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
 				//checker.cacheable = false
 				checker.hasLimit = true
-				checker.offsetAndCount = append(checker.offsetAndCount, count.GetInt64())
-				return in, false
+				countNum := count.GetInt64()
+				checker.offsetAndCount = append(checker.offsetAndCount, countNum)
+				// todo: check if > 10000 ---> cacheable
 			}
 		}
 		if node.Offset != nil {
@@ -474,7 +475,6 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 				//checker.cacheable = false
 				checker.hasLimit = true
 				checker.offsetAndCount = append(checker.offsetAndCount, offset.GetInt64())
-				return in, false
 			}
 		}
 	}
@@ -494,7 +494,7 @@ func getLimitFromAst(node ast.Node) []int64 {
 		//sctx:      sctx,
 		//schema:    is,
 		cacheable:      true,
-		offsetAndCount: make([]int64, 1),
+		offsetAndCount: []int64{},
 	}
 	node.Accept(&checker)
 	return checker.offsetAndCount

From 83f7a9dca907b6f00c130ef44b8ba8f990b3b77e Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 15:19:18 +0800
Subject: [PATCH 06/31] Update plan_cache_utils.go

---
 planner/core/plan_cache_utils.go | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index c8dcee84de336..d16fca8cd121f 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -285,7 +285,7 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 	if sessionVars.TimeZone != nil {
 		_, timezoneOffset = time.Now().In(sessionVars.TimeZone).Zone()
 	}
-	limit := getLimitFromAst(stmtNode)
+	offsetAndCount := getLimitFromAst(stmtNode)
 	key := &planCacheKey{
 		database:                 stmtDB,
 		connID:                   sessionVars.ConnectionID,
@@ -300,7 +300,7 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 		inRestrictedSQL:          sessionVars.InRestrictedSQL,
 		restrictedReadOnly:       variable.RestrictedReadOnly.Load(),
 		TiDBSuperReadOnly:        variable.VarTiDBSuperReadOnly.Load(),
-		limitOffsetAndCount:      limit,
+		limitOffsetAndCount:      offsetAndCount,
 	}
 	for k, v := range sessionVars.IsolationReadEngines {
 		key.isolationReadEngines[k] = v
@@ -450,8 +450,6 @@ func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCa
 }
 
 type limitExtractor struct {
-	//sctx           sessionctx.Context
-	//schema         infoschema.InfoSchema
 	cacheable      bool
 	hasLimit       bool
 	offsetAndCount []int64
@@ -491,8 +489,6 @@ func getLimitFromAst(node ast.Node) []int64 {
 		return []int64{}
 	}
 	checker := limitExtractor{
-		//sctx:      sctx,
-		//schema:    is,
 		cacheable:      true,
 		offsetAndCount: []int64{},
 	}

From b5f17a44e96d7680271d1aa725fcd467c2770bc8 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 16:17:29 +0800
Subject: [PATCH 07/31] safe value

---
 executor/prepared.go                   |  3 ++-
 planner/core/plan_cache.go             | 12 +++++++++---
 planner/core/plan_cache_utils.go       | 20 ++++++++------------
 planner/core/plan_cache_utils_test.go  |  8 ++++----
 planner/core/plan_cacheable_checker.go | 15 ---------------
 server/driver_tidb.go                  |  3 ++-
 session/session.go                     |  3 ++-
 7 files changed, 27 insertions(+), 37 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index 14b01d32ea3c2..00484ac525389 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -204,8 +204,9 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	delete(vars.PreparedStmtNameToID, e.Name)
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
+		limitOffsetAndCount, _ := plannercore.ExtractLimitFromAst(prepared.Stmt)
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL, prepared.Stmt)
+			0, bindSQL, limitOffsetAndCount)
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 13167ab4ce3a6..99a15f8443917 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -131,6 +131,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 
 	var bindSQL string
 	var ignorePlanCache = false
+	var canBeCached = true
 
 	// In rc or for update read, we need the latest schema version to decide whether we need to
 	// rebuild the plan. So we set this value in rc or for update read. In other cases, let it be 0.
@@ -144,8 +145,12 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			// up-to-date schema version which can lead plan cache miss and thus, the plan will be rebuilt.
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
+		var limitOffsetAndCount []int64
+		if limitOffsetAndCount, canBeCached = ExtractLimitFromAst(stmt.PreparedAst.Stmt); canBeCached {
+			sctx.GetSessionVars().StmtCtx.AppendWarning(errors.New("plan with limit count more than 10000 can't be cached"))
+		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst.Stmt); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitOffsetAndCount); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -165,7 +170,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 		}
 	}
 
-	return generateNewPlan(ctx, sctx, isNonPrepared, is, stmt, ignorePlanCache, cacheKey,
+	return generateNewPlan(ctx, sctx, isNonPrepared, is, stmt, ignorePlanCache && canBeCached, cacheKey,
 		latestSchemaVersion, paramNum, paramTypes, bindSQL)
 }
 
@@ -284,8 +289,9 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		// rebuild key to exclude kv.TiFlash when stmt is not read only
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
+			limitOffsetAndCount, _ := ExtractLimitFromAst(stmt.PreparedAst.Stmt)
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, stmt.PreparedAst.Stmt); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitOffsetAndCount); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index d16fca8cd121f..deb2903149d71 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -271,7 +271,7 @@ func SetPstmtIDSchemaVersion(key kvcache.Key, stmtText string, schemaVersion int
 // Note: lastUpdatedSchemaVersion will only be set in the case of rc or for update read in order to
 // differentiate the cache key. In other cases, it will be 0.
 func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string, schemaVersion int64,
-	lastUpdatedSchemaVersion int64, bindSQL string, stmtNode ast.StmtNode) (kvcache.Key, error) {
+	lastUpdatedSchemaVersion int64, bindSQL string, offsetAndCount []int64) (kvcache.Key, error) {
 	if stmtText == "" {
 		return nil, errors.New("no statement text")
 	}
@@ -285,7 +285,6 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 	if sessionVars.TimeZone != nil {
 		_, timezoneOffset = time.Now().In(sessionVars.TimeZone).Zone()
 	}
-	offsetAndCount := getLimitFromAst(stmtNode)
 	key := &planCacheKey{
 		database:                 stmtDB,
 		connID:                   sessionVars.ConnectionID,
@@ -450,8 +449,7 @@ func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCa
 }
 
 type limitExtractor struct {
-	cacheable      bool
-	hasLimit       bool
+	cacheable      bool // For safety considerations, check if limit count less than 10000
 	offsetAndCount []int64
 }
 
@@ -461,17 +459,15 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 	case *ast.Limit:
 		if node.Count != nil {
 			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-				//checker.cacheable = false
-				checker.hasLimit = true
 				countNum := count.GetInt64()
 				checker.offsetAndCount = append(checker.offsetAndCount, countNum)
-				// todo: check if > 10000 ---> cacheable
+				if countNum > 10000 {
+					checker.cacheable = false
+				}
 			}
 		}
 		if node.Offset != nil {
 			if offset, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
-				//checker.cacheable = false
-				checker.hasLimit = true
 				checker.offsetAndCount = append(checker.offsetAndCount, offset.GetInt64())
 			}
 		}
@@ -484,14 +480,14 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 	return in, checker.cacheable
 }
 
-func getLimitFromAst(node ast.Node) []int64 {
+func ExtractLimitFromAst(node ast.Node) ([]int64, bool) {
 	if node == nil {
-		return []int64{}
+		return []int64{}, true
 	}
 	checker := limitExtractor{
 		cacheable:      true,
 		offsetAndCount: []int64{},
 	}
 	node.Accept(&checker)
-	return checker.offsetAndCount
+	return checker.offsetAndCount, checker.cacheable
 }
diff --git a/planner/core/plan_cache_utils_test.go b/planner/core/plan_cache_utils_test.go
index 2ef6766231312..55662defe90a7 100644
--- a/planner/core/plan_cache_utils_test.go
+++ b/planner/core/plan_cache_utils_test.go
@@ -32,20 +32,20 @@ func TestCacheKey(t *testing.T) {
 	ctx.GetSessionVars().InRestrictedSQL = false
 	variable.RestrictedReadOnly.Store(false)
 	variable.VarTiDBSuperReadOnly.Store(false)
-	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", nil)
+	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", []int64{})
 	if err.Error() != "no statement text" {
 		t.Fail() // no statement text
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", nil)
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", []int64{})
 	if err != nil {
 		t.Fail() // schema can be nil
 	}
 	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1,
-		"select /*+ ignore_plan_cache() */ * from t", nil)
+		"select /*+ ignore_plan_cache() */ * from t", []int64{})
 	if err != nil {
 		t.Fail()
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", nil)
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", []int64{})
 	if err != nil {
 		t.Fail()
 	}
diff --git a/planner/core/plan_cacheable_checker.go b/planner/core/plan_cacheable_checker.go
index 3eca17c1708e7..fedcdfe59c151 100644
--- a/planner/core/plan_cacheable_checker.go
+++ b/planner/core/plan_cacheable_checker.go
@@ -135,21 +135,6 @@ func (checker *cacheableChecker) Enter(in ast.Node) (out ast.Node, skipChildren
 				return in, true
 			}
 		}
-	//case *ast.Limit:
-	//	if node.Count != nil {
-	//		if _, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-	//			checker.cacheable = false
-	//			checker.reason = "query has 'limit ?' is un-cacheable"
-	//			return in, true
-	//		}
-	//	}
-	//	if node.Offset != nil {
-	//		if _, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
-	//			checker.cacheable = false
-	//			checker.reason = "query has 'limit ?, 10' is un-cacheable"
-	//			return in, true
-	//		}
-	//	}
 	case *ast.FrameBound:
 		if _, ok := node.Expr.(*driver.ParamMarkerExpr); ok {
 			checker.cacheable = false
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index ac983edc50585..bf9a4374420f1 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -171,8 +171,9 @@ func (ts *TiDBStatement) Close() error {
 				return errors.Errorf("invalid PlanCacheStmt type")
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
+			limitOffsetAndCount, _ := core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt)
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, preparedObj.PreparedAst.Stmt)
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, limitOffsetAndCount)
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index 154337b053211..81b0f16db8cd2 100644
--- a/session/session.go
+++ b/session/session.go
@@ -382,8 +382,9 @@ func (s *session) cleanRetryInfo() {
 				preparedAst = preparedObj.PreparedAst
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
+				limitOffsetAndCount, _ := plannercore.ExtractLimitFromAst(preparedAst.Stmt)
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL, preparedAst.Stmt)
+					0, bindSQL, limitOffsetAndCount)
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return

From a7a5844f8db301415f2b394534f72e3bf1a909c7 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 17:02:42 +0800
Subject: [PATCH 08/31] fix

---
 planner/core/plan_cache.go       | 2 +-
 planner/core/plan_cache_utils.go | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 99a15f8443917..8ab919ffb1770 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -146,7 +146,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
 		var limitOffsetAndCount []int64
-		if limitOffsetAndCount, canBeCached = ExtractLimitFromAst(stmt.PreparedAst.Stmt); canBeCached {
+		if limitOffsetAndCount, canBeCached = ExtractLimitFromAst(stmt.PreparedAst.Stmt); !canBeCached {
 			sctx.GetSessionVars().StmtCtx.AppendWarning(errors.New("plan with limit count more than 10000 can't be cached"))
 		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index deb2903149d71..16f16ba96373d 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -480,6 +480,7 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 	return in, checker.cacheable
 }
 
+// ExtractLimitFromAst extract limit offset and count from ast for plan cache key encode
 func ExtractLimitFromAst(node ast.Node) ([]int64, bool) {
 	if node == nil {
 		return []int64{}, true

From 6188f82112dcb80bf594692e3c02a1a7b798c394 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 28 Dec 2022 17:46:02 +0800
Subject: [PATCH 09/31] Update plan_cache_test.go

---
 planner/core/plan_cache_test.go | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index 7479c7a70718e..011dc2288ba7e 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -315,7 +315,6 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		{"prepare stmt from 'insert into t select * from t order by a desc limit ?, ?'", []int{1, 2}},
 		{"prepare stmt from 'update t set a = 1 limit ?'", []int{1}},
 		{" prepare stmt from '(select * from t order by a limit ?) union (select * from t order by a desc limit ?)';", []int{1, 2}},
-		{" prepare stmt from 'select * from t where a in (select b from t limit ?) limit ?;';", []int{10, 5}},
 	}
 
 	for _, testCase := range testCases {
@@ -335,4 +334,10 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
 	}
+
+	// error case
+	tk.MustExec("prepare stmt from 'select * from t limit ?'")
+	tk.MustExec("set @a = 10001")
+	tk.MustExec("execute stmt using @a")
+	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 plan with limit count more than 10000 can't be cached"))
 }

From fbe23e52a1f8ad98822aa9e7c34d2b3fdf42307f Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 3 Jan 2023 10:35:09 +0800
Subject: [PATCH 10/31] unify error message

---
 planner/core/plan_cache.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index cd94bfc2dacd9..88e059fc46e03 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -147,7 +147,7 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 		}
 		var limitOffsetAndCount []int64
 		if limitOffsetAndCount, canBeCached = ExtractLimitFromAst(stmt.PreparedAst.Stmt); !canBeCached {
-			sctx.GetSessionVars().StmtCtx.AppendWarning(errors.New("plan with limit count more than 10000 can't be cached"))
+			sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: limit count more than 10000"))
 		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
 			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitOffsetAndCount); err != nil {

From 190bc29f570052b678015f0f29d5e2f8d9c23199 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 3 Jan 2023 18:20:22 +0800
Subject: [PATCH 11/31] check limit argument

---
 planner/core/logical_plan_builder.go | 62 +++++++++++++++++++++++++++-
 1 file changed, 60 insertions(+), 2 deletions(-)

diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index 592bb55f79619..126936274c473 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -2041,17 +2041,75 @@ func getUintFromNode(ctx sessionctx.Context, n ast.Node) (uVal uint64, isNull bo
 	return 0, false, false
 }
 
+// getUintFromLimitNode gets uint64 value from ast.Node.
+// For ordinary statement, node should be uint64 constant value.
+// For prepared statement, node is string. We should convert it to uint64.
+func getUintFromLimitNode(ctx sessionctx.Context, n ast.Node) (uVal uint64, isNull bool, isExpectedType bool) {
+	var val interface{}
+	switch v := n.(type) {
+	case *driver.ValueExpr:
+		val = v.GetValue()
+	case *driver.ParamMarkerExpr:
+		if !v.InExecute {
+			return 0, false, true
+		}
+		param, err := expression.ParamMarkerExpression(ctx, v, false)
+		if err != nil {
+			return 0, false, false
+		}
+		str, isNull, err := expression.GetStringFromConstant(ctx, param)
+		if err != nil {
+			return 0, false, false
+		}
+		if isNull {
+			return 0, true, true
+		}
+		if ok := isAllDigit(str); !ok {
+			return 0, false, false
+		}
+		val = str
+	default:
+		return 0, false, false
+	}
+	switch v := val.(type) {
+	case uint64:
+		return v, false, true
+	case int64:
+		if v >= 0 {
+			return uint64(v), false, true
+		}
+	case string:
+		sc := ctx.GetSessionVars().StmtCtx
+		uVal, err := types.StrToUint(sc, v, false)
+		if err != nil {
+			return 0, false, false
+		}
+		return uVal, false, true
+	}
+	return 0, false, false
+}
+
+func isAllDigit(str string) bool {
+	for _, c := range str {
+		if c < '0' || c > '9' {
+			return false
+		}
+	}
+	return true
+}
+
 func extractLimitCountOffset(ctx sessionctx.Context, limit *ast.Limit) (count uint64,
 	offset uint64, err error) {
 	var isExpectedType bool
 	if limit.Count != nil {
-		count, _, isExpectedType = getUintFromNode(ctx, limit.Count)
+		count, _, isExpectedType = getUintFromLimitNode(ctx, limit.Count)
 		if !isExpectedType {
 			return 0, 0, ErrWrongArguments.GenWithStackByArgs("LIMIT")
 		}
 	}
+
 	if limit.Offset != nil {
-		offset, _, isExpectedType = getUintFromNode(ctx, limit.Offset)
+		offset, _, isExpectedType = getUintFromLimitNode(ctx, limit.Offset)
 		if !isExpectedType {
 			return 0, 0, ErrWrongArguments.GenWithStackByArgs("LIMIT")
 		}

From f1f73db27dd067651c903179d1ece5a900a7b08f Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 3 Jan 2023 20:53:34 +0800
Subject: [PATCH 12/31] unify warning message

---
 executor/prepared.go             | 3 +--
 planner/core/plan_cache.go       | 9 ++-------
 planner/core/plan_cache_utils.go | 9 ++++++---
 server/driver_tidb.go            | 3 +--
 session/session.go               | 3 +--
 5 files changed, 11 insertions(+), 16 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index 00484ac525389..75d3e485dc3ea 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -204,9 +204,8 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	delete(vars.PreparedStmtNameToID, e.Name)
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
-		limitOffsetAndCount, _ := plannercore.ExtractLimitFromAst(prepared.Stmt)
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL, limitOffsetAndCount)
+			0, bindSQL, plannercore.ExtractLimitFromAst(prepared.Stmt, nil))
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 88e059fc46e03..4b5060012da5f 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -145,12 +145,8 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			// up-to-date schema version which can lead plan cache miss and thus, the plan will be rebuilt.
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
-		var limitOffsetAndCount []int64
-		if limitOffsetAndCount, canBeCached = ExtractLimitFromAst(stmt.PreparedAst.Stmt); !canBeCached {
-			sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: limit count more than 10000"))
-		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitOffsetAndCount); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -289,9 +285,8 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		// rebuild key to exclude kv.TiFlash when stmt is not read only
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
-			limitOffsetAndCount, _ := ExtractLimitFromAst(stmt.PreparedAst.Stmt)
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitOffsetAndCount); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 3783cae566481..dca9334ecc286 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -484,14 +484,17 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 }
 
 // ExtractLimitFromAst extract limit offset and count from ast for plan cache key encode
-func ExtractLimitFromAst(node ast.Node) ([]int64, bool) {
+func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) []int64 {
 	if node == nil {
-		return []int64{}, true
+		return []int64{}
 	}
 	checker := limitExtractor{
 		cacheable:      true,
 		offsetAndCount: []int64{},
 	}
 	node.Accept(&checker)
-	return checker.offsetAndCount, checker.cacheable
+	if sctx != nil && !checker.cacheable {
+		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: limit count more than 10000"))
+	}
+	return checker.offsetAndCount
 }
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index bf9a4374420f1..26b3ae180dc4b 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -171,9 +171,8 @@ func (ts *TiDBStatement) Close() error {
 				return errors.Errorf("invalid PlanCacheStmt type")
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
-			limitOffsetAndCount, _ := core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt)
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, limitOffsetAndCount)
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt, nil))
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index 15e6cad8520db..5d96fc66082fe 100644
--- a/session/session.go
+++ b/session/session.go
@@ -383,9 +383,8 @@ func (s *session) cleanRetryInfo() {
 				preparedAst = preparedObj.PreparedAst
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
-				limitOffsetAndCount, _ := plannercore.ExtractLimitFromAst(preparedAst.Stmt)
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL, limitOffsetAndCount)
+					0, bindSQL, plannercore.ExtractLimitFromAst(preparedAst.Stmt, nil))
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return

From ed829d1d6a5d498499bd76150a1ada84a7ad12f5 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Tue, 3 Jan 2023 22:41:01 +0800
Subject: [PATCH 13/31] fix ut

---
 executor/seqtest/prepared_test.go |  6 +++---
 planner/core/plan_cache_test.go   | 18 +-----------------
 2 files changed, 4 insertions(+), 20 deletions(-)

diff --git a/executor/seqtest/prepared_test.go b/executor/seqtest/prepared_test.go
index 5edbed52b4e13..f3c81522be83a 100644
--- a/executor/seqtest/prepared_test.go
+++ b/executor/seqtest/prepared_test.go
@@ -280,11 +280,11 @@ func TestPreparedLimitOffset(t *testing.T) {
 		r.Check(testkit.Rows("2"))
 
 		tk.MustExec(`set @a=1.1`)
-		r = tk.MustQuery(`execute stmt_test_1 using @a, @b;`)
-		r.Check(testkit.Rows("2"))
+		_, err := tk.Exec(`execute stmt_test_1 using @a, @b;`)
+		require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		tk.MustExec(`set @c="-1"`)
-		_, err := tk.Exec("execute stmt_test_1 using @c, @c")
+		_, err = tk.Exec("execute stmt_test_1 using @c, @c")
 		require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		stmtID, _, _, err := tk.Session().PrepareStmt("select id from prepare_test limit ?")
diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index 11a6217dd102a..2d24b998c1c18 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -379,22 +379,6 @@ func TestPlanCacheDiagInfo(t *testing.T) {
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: some parameters may be overwritten"))
 }
 
-func TestUncacheableReason(t *testing.T) {
-	store := testkit.CreateMockStore(t)
-	tk := testkit.NewTestKit(t, store)
-	tk.MustExec("use test")
-	tk.MustExec("create table t (a int)")
-
-	tk.MustExec("prepare st from 'select * from t limit ?'")
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'limit ?' is un-cacheable"))
-
-	tk.MustExec("set @a=1")
-	tk.MustQuery("execute st using @a").Check(testkit.Rows())
-	tk.MustExec("prepare st from 'select * from t limit ?'")
-	// show the corresponding un-cacheable reason at execute-stage as well
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: query has 'limit ?' is un-cacheable"))
-}
-
 func TestPlanCacheWithLimit(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
@@ -437,5 +421,5 @@ func TestPlanCacheWithLimit(t *testing.T) {
 	tk.MustExec("prepare stmt from 'select * from t limit ?'")
 	tk.MustExec("set @a = 10001")
 	tk.MustExec("execute stmt using @a")
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 plan with limit count more than 10000 can't be cached"))
+	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: limit count more than 10000"))
 }

From 8244f14adad3e76642fb68eb6ec3cde6ef88ded4 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 4 Jan 2023 15:58:50 +0800
Subject: [PATCH 14/31] revert

---
 executor/seqtest/prepared_test.go    |  6 +--
 planner/core/logical_plan_builder.go | 62 +---------------------------
 2 files changed, 5 insertions(+), 63 deletions(-)

diff --git a/executor/seqtest/prepared_test.go b/executor/seqtest/prepared_test.go
index f3c81522be83a..5edbed52b4e13 100644
--- a/executor/seqtest/prepared_test.go
+++ b/executor/seqtest/prepared_test.go
@@ -280,11 +280,11 @@ func TestPreparedLimitOffset(t *testing.T) {
 		r.Check(testkit.Rows("2"))
 
 		tk.MustExec(`set @a=1.1`)
-		_, err := tk.Exec(`execute stmt_test_1 using @a, @b;`)
-		require.True(t, plannercore.ErrWrongArguments.Equal(err))
+		r = tk.MustQuery(`execute stmt_test_1 using @a, @b;`)
+		r.Check(testkit.Rows("2"))
 
 		tk.MustExec(`set @c="-1"`)
-		_, err = tk.Exec("execute stmt_test_1 using @c, @c")
+		_, err := tk.Exec("execute stmt_test_1 using @c, @c")
 		require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		stmtID, _, _, err := tk.Session().PrepareStmt("select id from prepare_test limit ?")
diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index 126936274c473..592bb55f79619 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -2041,75 +2041,17 @@ func getUintFromNode(ctx sessionctx.Context, n ast.Node) (uVal uint64, isNull bo
 	return 0, false, false
 }
 
-// getUintFromLimitNode gets uint64 value from ast.Node.
-// For ordinary statement, node should be uint64 constant value.
-// For prepared statement, node is string. We should convert it to uint64.
-func getUintFromLimitNode(ctx sessionctx.Context, n ast.Node) (uVal uint64, isNull bool, isExpectedType bool) {
-	var val interface{}
-	switch v := n.(type) {
-	case *driver.ValueExpr:
-		val = v.GetValue()
-	case *driver.ParamMarkerExpr:
-		if !v.InExecute {
-			return 0, false, true
-		}
-		param, err := expression.ParamMarkerExpression(ctx, v, false)
-		if err != nil {
-			return 0, false, false
-		}
-		str, isNull, err := expression.GetStringFromConstant(ctx, param)
-		if err != nil {
-			return 0, false, false
-		}
-		if isNull {
-			return 0, true, true
-		}
-		if ok := isAllDigit(str); !ok {
-			return 0, false, false
-		}
-		val = str
-	default:
-		return 0, false, false
-	}
-	switch v := val.(type) {
-	case uint64:
-		return v, false, true
-	case int64:
-		if v >= 0 {
-			return uint64(v), false, true
-		}
-	case string:
-		sc := ctx.GetSessionVars().StmtCtx
-		uVal, err := types.StrToUint(sc, v, false)
-		if err != nil {
-			return 0, false, false
-		}
-		return uVal, false, true
-	}
-	return 0, false, false
-}
-
-func isAllDigit(str string) bool {
-	for _, c := range str {
-		if c < '0' || c > '9' {
-			return false
-		}
-	}
-	return true
-}
-
 func extractLimitCountOffset(ctx sessionctx.Context, limit *ast.Limit) (count uint64,
 	offset uint64, err error) {
 	var isExpectedType bool
 	if limit.Count != nil {
-		count, _, isExpectedType = getUintFromLimitNode(ctx, limit.Count)
+		count, _, isExpectedType = getUintFromNode(ctx, limit.Count)
 		if !isExpectedType {
 			return 0, 0, ErrWrongArguments.GenWithStackByArgs("LIMIT")
 		}
 	}
-
 	if limit.Offset != nil {
-		offset, _, isExpectedType = getUintFromLimitNode(ctx, limit.Offset)
+		offset, _, isExpectedType = getUintFromNode(ctx, limit.Offset)
 		if !isExpectedType {
 			return 0, 0, ErrWrongArguments.GenWithStackByArgs("LIMIT")
 		}

From 8258da04445bc92a6de18ab324d2463c6426a521 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 4 Jan 2023 17:26:26 +0800
Subject: [PATCH 15/31] only_int

---
 planner/core/plan_cache_utils.go | 41 ++++++++++++++++++++++++++------
 1 file changed, 34 insertions(+), 7 deletions(-)

diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 4f4fe40c9d3a8..ce9a9c51b3b0b 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -460,8 +460,9 @@ func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCa
 }
 
 type limitExtractor struct {
-	cacheable      bool // For safety considerations, check if limit count less than 10000
-	offsetAndCount []int64
+	cacheable         bool // For safety considerations, check if limit count less than 10000
+	offsetAndCount    []int64
+	unCacheableReason string
 }
 
 // Enter implements Visitor interface.
@@ -470,16 +471,33 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 	case *ast.Limit:
 		if node.Count != nil {
 			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-				countNum := count.GetInt64()
-				checker.offsetAndCount = append(checker.offsetAndCount, countNum)
-				if countNum > 10000 {
+				// currently, we just support INT type parameters, eg: set @a = 123
+				typeExpected, val := checkLimitParamType(count)
+				if typeExpected {
+					if val > 10000 {
+						checker.cacheable = false
+						checker.unCacheableReason = "limit count more than 10000"
+						return in, true
+					} else {
+						checker.offsetAndCount = append(checker.offsetAndCount, val)
+					}
+				} else {
 					checker.cacheable = false
+					checker.unCacheableReason = "limit count type un-cacheable"
+					return in, true
 				}
 			}
 		}
 		if node.Offset != nil {
 			if offset, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
-				checker.offsetAndCount = append(checker.offsetAndCount, offset.GetInt64())
+				typeExpected, val := checkLimitParamType(offset)
+				if typeExpected {
+					checker.offsetAndCount = append(checker.offsetAndCount, val)
+				} else {
+					checker.cacheable = false
+					checker.unCacheableReason = "limit offset type un-cacheable"
+					return in, true
+				}
 			}
 		}
 	}
@@ -502,7 +520,16 @@ func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) []int64 {
 	}
 	node.Accept(&checker)
 	if sctx != nil && !checker.cacheable {
-		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: limit count more than 10000"))
+		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: " + checker.unCacheableReason))
 	}
 	return checker.offsetAndCount
 }
+
+func checkLimitParamType(node *driver.ParamMarkerExpr) (bool, int64) {
+	val := node.GetValue()
+	switch v := val.(type) {
+	case int64:
+		return true, v
+	}
+	return false, -1
+}

From 51d06a4472d0c80cea98d0e774a14f9641ecf387 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 4 Jan 2023 17:33:12 +0800
Subject: [PATCH 16/31] Update plan_cache_utils.go

---
 planner/core/plan_cache_utils.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index ce9a9c51b3b0b..699010be67123 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -478,9 +478,8 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 						checker.cacheable = false
 						checker.unCacheableReason = "limit count more than 10000"
 						return in, true
-					} else {
-						checker.offsetAndCount = append(checker.offsetAndCount, val)
 					}
+					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
 					checker.cacheable = false
 					checker.unCacheableReason = "limit count type un-cacheable"

From 5d83ceaf6e45006a91f6df2de5aa3af73d583a48 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Thu, 5 Jan 2023 17:34:11 +0800
Subject: [PATCH 17/31] only int in limit stmt

---
 executor/prepared.go             |  6 +++++-
 planner/core/plan_cache.go       | 12 ++++++++++--
 planner/core/plan_cache_test.go  | 28 +++++++++++++++++++++++++++-
 planner/core/plan_cache_utils.go | 16 +++++++++-------
 server/driver_tidb.go            |  6 +++++-
 session/session.go               |  7 ++++++-
 6 files changed, 62 insertions(+), 13 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index 75d3e485dc3ea..a93adcb3d6368 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -204,8 +204,12 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	delete(vars.PreparedStmtNameToID, e.Name)
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
+		limitCountAndOffset, paramErr := plannercore.ExtractLimitFromAst(prepared.Stmt, nil)
+		if paramErr != nil {
+			return paramErr
+		}
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL, plannercore.ExtractLimitFromAst(prepared.Stmt, nil))
+			0, bindSQL, limitCountAndOffset)
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 8da12cbe89026..5a9f6f92905e1 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -152,8 +152,12 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			// up-to-date schema version which can lead plan cache miss and thus, the plan will be rebuilt.
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
+		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)
+		if paramErr != nil {
+			return nil, nil, paramErr
+		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitCountAndOffset); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -290,8 +294,12 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		// rebuild key to exclude kv.TiFlash when stmt is not read only
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
+			limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)
+			if paramErr != nil {
+				return nil, nil, paramErr
+			}
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitCountAndOffset); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index 7dd7f2d56a9db..ab48a08019710 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -442,10 +442,36 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
 	}
+}
 
-	// error case
+func TestUnsupportedLimitCase(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int, key(a))")
 	tk.MustExec("prepare stmt from 'select * from t limit ?'")
+
 	tk.MustExec("set @a = 10001")
 	tk.MustExec("execute stmt using @a")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: limit count more than 10000"))
+
+	tk.MustExec("set @a = 0")
+	tk.MustExec("execute stmt using @a")
+	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: get a TableDual plan"))
+
+	tk.MustExec("set @a = 1.2")
+	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
+
+	tk.MustExec("set @a = 1.")
+	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
+
+	tk.MustExec("set @a = '0'")
+	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
+
+	tk.MustExec("set @a = '1'")
+	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
+
+	tk.MustExec("set @a = 1_2")
+	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
 }
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 699010be67123..29fc6bfc694a0 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -463,6 +463,7 @@ type limitExtractor struct {
 	cacheable         bool // For safety considerations, check if limit count less than 10000
 	offsetAndCount    []int64
 	unCacheableReason string
+	paramTypeErr      error
 }
 
 // Enter implements Visitor interface.
@@ -481,8 +482,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 					}
 					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
-					checker.cacheable = false
-					checker.unCacheableReason = "limit count type un-cacheable"
+					checker.paramTypeErr = errors.New("incorrect arguments to limit in plan cache")
 					return in, true
 				}
 			}
@@ -493,8 +493,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 				if typeExpected {
 					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
-					checker.cacheable = false
-					checker.unCacheableReason = "limit offset type un-cacheable"
+					checker.paramTypeErr = errors.New("incorrect arguments to limit in plan cache")
 					return in, true
 				}
 			}
@@ -509,19 +508,22 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 }
 
 // ExtractLimitFromAst extract limit offset and count from ast for plan cache key encode
-func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) []int64 {
+func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]int64, error) {
 	if node == nil {
-		return []int64{}
+		return []int64{}, nil
 	}
 	checker := limitExtractor{
 		cacheable:      true,
 		offsetAndCount: []int64{},
 	}
 	node.Accept(&checker)
+	if checker.paramTypeErr != nil {
+		return []int64{}, checker.paramTypeErr
+	}
 	if sctx != nil && !checker.cacheable {
 		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: " + checker.unCacheableReason))
 	}
-	return checker.offsetAndCount
+	return checker.offsetAndCount, nil
 }
 
 func checkLimitParamType(node *driver.ParamMarkerExpr) (bool, int64) {
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index 26b3ae180dc4b..a822fa014ef2c 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -171,8 +171,12 @@ func (ts *TiDBStatement) Close() error {
 				return errors.Errorf("invalid PlanCacheStmt type")
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
+			limitCountAndOffset, paramErr := core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt, nil)
+			if paramErr != nil {
+				return paramErr
+			}
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt, nil))
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, limitCountAndOffset)
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index e67e3eb6f4d7f..a6770fa24502e 100644
--- a/session/session.go
+++ b/session/session.go
@@ -383,8 +383,13 @@ func (s *session) cleanRetryInfo() {
 				preparedAst = preparedObj.PreparedAst
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
+				limitCountAndOffset, paramErr := plannercore.ExtractLimitFromAst(preparedAst.Stmt, nil)
+				if paramErr != nil {
+					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(paramErr))
+					return
+				}
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL, plannercore.ExtractLimitFromAst(preparedAst.Stmt, nil))
+					0, bindSQL, limitCountAndOffset)
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return

From aae94e15fc4606373f911076a83c8f61ea7c32dd Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Fri, 6 Jan 2023 11:18:07 +0800
Subject: [PATCH 18/31] fix

---
 planner/core/plan_cache_test.go | 26 --------------------------
 1 file changed, 26 deletions(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index ab48a08019710..804256aa66bfb 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -442,36 +442,10 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
 	}
-}
 
-func TestUnsupportedLimitCase(t *testing.T) {
-	store := testkit.CreateMockStore(t)
-	tk := testkit.NewTestKit(t, store)
-	tk.MustExec("use test")
-	tk.MustExec("drop table if exists t")
-	tk.MustExec("create table t(a int, key(a))")
 	tk.MustExec("prepare stmt from 'select * from t limit ?'")
-
 	tk.MustExec("set @a = 10001")
 	tk.MustExec("execute stmt using @a")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: limit count more than 10000"))
 
-	tk.MustExec("set @a = 0")
-	tk.MustExec("execute stmt using @a")
-	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: get a TableDual plan"))
-
-	tk.MustExec("set @a = 1.2")
-	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
-
-	tk.MustExec("set @a = 1.")
-	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
-
-	tk.MustExec("set @a = '0'")
-	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
-
-	tk.MustExec("set @a = '1'")
-	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
-
-	tk.MustExec("set @a = 1_2")
-	tk.MustGetErrMsg("execute stmt using @a", "incorrect arguments to limit in plan cache")
 }

From 0d6335a2ddc4ac24d01aad77936974437b2a7c6d Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 14:59:50 +0800
Subject: [PATCH 19/31] replace int by uint in cache key

---
 planner/core/plan_cache_utils.go | 23 ++++++++++++-----------
 1 file changed, 12 insertions(+), 11 deletions(-)

diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 29fc6bfc694a0..055e5b9700f10 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -194,7 +194,7 @@ type planCacheKey struct {
 	inRestrictedSQL          bool
 	restrictedReadOnly       bool
 	TiDBSuperReadOnly        bool
-	limitOffsetAndCount      []int64
+	limitOffsetAndCount      []uint64
 
 	memoryUsage int64 // Do not include in hash
 	hash        []byte
@@ -232,7 +232,7 @@ func (key *planCacheKey) Hash() []byte {
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.restrictedReadOnly))...)
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.TiDBSuperReadOnly))...)
 		for _, l := range key.limitOffsetAndCount {
-			key.hash = codec.EncodeInt(key.hash, l)
+			key.hash = codec.EncodeUint(key.hash, l)
 		}
 	}
 	return key.hash
@@ -275,7 +275,7 @@ func SetPstmtIDSchemaVersion(key kvcache.Key, stmtText string, schemaVersion int
 // Note: lastUpdatedSchemaVersion will only be set in the case of rc or for update read in order to
 // differentiate the cache key. In other cases, it will be 0.
 func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string, schemaVersion int64,
-	lastUpdatedSchemaVersion int64, bindSQL string, offsetAndCount []int64) (kvcache.Key, error) {
+	lastUpdatedSchemaVersion int64, bindSQL string, offsetAndCount []uint64) (kvcache.Key, error) {
 	if stmtText == "" {
 		return nil, errors.New("no statement text")
 	}
@@ -461,7 +461,7 @@ func GetPreparedStmt(stmt *ast.ExecuteStmt, vars *variable.SessionVars) (*PlanCa
 
 type limitExtractor struct {
 	cacheable         bool // For safety considerations, check if limit count less than 10000
-	offsetAndCount    []int64
+	offsetAndCount    []uint64
 	unCacheableReason string
 	paramTypeErr      error
 }
@@ -472,7 +472,6 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 	case *ast.Limit:
 		if node.Count != nil {
 			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-				// currently, we just support INT type parameters, eg: set @a = 123
 				typeExpected, val := checkLimitParamType(count)
 				if typeExpected {
 					if val > 10000 {
@@ -508,17 +507,17 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 }
 
 // ExtractLimitFromAst extract limit offset and count from ast for plan cache key encode
-func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]int64, error) {
+func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]uint64, error) {
 	if node == nil {
-		return []int64{}, nil
+		return []uint64{}, nil
 	}
 	checker := limitExtractor{
 		cacheable:      true,
-		offsetAndCount: []int64{},
+		offsetAndCount: []uint64{},
 	}
 	node.Accept(&checker)
 	if checker.paramTypeErr != nil {
-		return []int64{}, checker.paramTypeErr
+		return []uint64{}, checker.paramTypeErr
 	}
 	if sctx != nil && !checker.cacheable {
 		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: " + checker.unCacheableReason))
@@ -526,11 +525,13 @@ func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]int64, error
 	return checker.offsetAndCount, nil
 }
 
-func checkLimitParamType(node *driver.ParamMarkerExpr) (bool, int64) {
+func checkLimitParamType(node *driver.ParamMarkerExpr) (bool, uint64) {
 	val := node.GetValue()
 	switch v := val.(type) {
 	case int64:
+		return true, uint64(v)
+	case uint64:
 		return true, v
 	}
-	return false, -1
+	return false, 0
 }

From 4305ba1b670a4b4e1d91b3bff4623b8e48abfba4 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 15:14:51 +0800
Subject: [PATCH 20/31] Update plan_cache_utils_test.go

---
 planner/core/plan_cache_utils_test.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/planner/core/plan_cache_utils_test.go b/planner/core/plan_cache_utils_test.go
index 55662defe90a7..b1ac33f56e6ba 100644
--- a/planner/core/plan_cache_utils_test.go
+++ b/planner/core/plan_cache_utils_test.go
@@ -32,20 +32,20 @@ func TestCacheKey(t *testing.T) {
 	ctx.GetSessionVars().InRestrictedSQL = false
 	variable.RestrictedReadOnly.Store(false)
 	variable.VarTiDBSuperReadOnly.Store(false)
-	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", []int64{})
+	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", []uint64{})
 	if err.Error() != "no statement text" {
 		t.Fail() // no statement text
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", []int64{})
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", []uint64{})
 	if err != nil {
 		t.Fail() // schema can be nil
 	}
 	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1,
-		"select /*+ ignore_plan_cache() */ * from t", []int64{})
+		"select /*+ ignore_plan_cache() */ * from t", []uint64{})
 	if err != nil {
 		t.Fail()
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", []int64{})
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", []uint64{})
 	if err != nil {
 		t.Fail()
 	}

From 61b15471242d010ec19a4bc2b8b5c16e5de43cdd Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 15:50:38 +0800
Subject: [PATCH 21/31] Update prepared_test.go

---
 executor/seqtest/prepared_test.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/executor/seqtest/prepared_test.go b/executor/seqtest/prepared_test.go
index 5edbed52b4e13..f3c81522be83a 100644
--- a/executor/seqtest/prepared_test.go
+++ b/executor/seqtest/prepared_test.go
@@ -280,11 +280,11 @@ func TestPreparedLimitOffset(t *testing.T) {
 		r.Check(testkit.Rows("2"))
 
 		tk.MustExec(`set @a=1.1`)
-		r = tk.MustQuery(`execute stmt_test_1 using @a, @b;`)
-		r.Check(testkit.Rows("2"))
+		_, err := tk.Exec(`execute stmt_test_1 using @a, @b;`)
+		require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		tk.MustExec(`set @c="-1"`)
-		_, err := tk.Exec("execute stmt_test_1 using @c, @c")
+		_, err = tk.Exec("execute stmt_test_1 using @c, @c")
 		require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		stmtID, _, _, err := tk.Session().PrepareStmt("select id from prepare_test limit ?")

From cd6ca23dec5d3ebcc6c3ed0605bb470bce31b866 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 15:54:48 +0800
Subject: [PATCH 22/31] Update prepared_test.go

---
 executor/seqtest/prepared_test.go | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/executor/seqtest/prepared_test.go b/executor/seqtest/prepared_test.go
index f3c81522be83a..376e517754133 100644
--- a/executor/seqtest/prepared_test.go
+++ b/executor/seqtest/prepared_test.go
@@ -279,13 +279,13 @@ func TestPreparedLimitOffset(t *testing.T) {
 		r := tk.MustQuery(`execute stmt_test_1 using @a, @b;`)
 		r.Check(testkit.Rows("2"))
 
-		tk.MustExec(`set @a=1.1`)
-		_, err := tk.Exec(`execute stmt_test_1 using @a, @b;`)
-		require.True(t, plannercore.ErrWrongArguments.Equal(err))
-
-		tk.MustExec(`set @c="-1"`)
-		_, err = tk.Exec("execute stmt_test_1 using @c, @c")
-		require.True(t, plannercore.ErrWrongArguments.Equal(err))
+		//tk.MustExec(`set @a=1.1`)
+		//_, err := tk.Exec(`execute stmt_test_1 using @a, @b;`)
+		//require.True(t, plannercore.ErrWrongArguments.Equal(err))
+		//
+		//tk.MustExec(`set @c="-1"`)
+		//_, err = tk.Exec("execute stmt_test_1 using @c, @c")
+		//require.True(t, plannercore.ErrWrongArguments.Equal(err))
 
 		stmtID, _, _, err := tk.Session().PrepareStmt("select id from prepare_test limit ?")
 		require.NoError(t, err)

From e921a81e03ce08ea92443ecbf32086c27f2a0f61 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 17:03:32 +0800
Subject: [PATCH 23/31] Update plan_cache_test.go

---
 planner/core/plan_cache_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index 629065ef8d562..f8310623e0c36 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -479,5 +479,4 @@ func TestPlanCacheWithLimit(t *testing.T) {
 	tk.MustExec("set @a = 10001")
 	tk.MustExec("execute stmt using @a")
 	tk.MustQuery("show warnings").Check(testkit.Rows("Warning 1105 skip plan-cache: limit count more than 10000"))
-
 }

From 9c4df35937d981d1d447792f21df30e979d15a11 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 9 Jan 2023 17:11:24 +0800
Subject: [PATCH 24/31] Update plan_cache_utils.go

---
 planner/core/plan_cache_utils.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 055e5b9700f10..7c72762394dc4 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -481,7 +481,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 					}
 					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
-					checker.paramTypeErr = errors.New("incorrect arguments to limit in plan cache")
+					checker.paramTypeErr = ErrWrongArguments.GenWithStackByArgs("LIMIT")
 					return in, true
 				}
 			}
@@ -492,7 +492,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 				if typeExpected {
 					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
-					checker.paramTypeErr = errors.New("incorrect arguments to limit in plan cache")
+					checker.paramTypeErr = ErrWrongArguments.GenWithStackByArgs("LIMIT")
 					return in, true
 				}
 			}

From 9dc74fb8d7df2b1d77176595c2147878bf734708 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Wed, 11 Jan 2023 20:06:23 +0800
Subject: [PATCH 25/31] use exist function

---
 planner/core/logical_plan_builder.go | 12 ++++++------
 planner/core/plan_cache_utils.go     | 15 ++-------------
 2 files changed, 8 insertions(+), 19 deletions(-)

diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index 619b95474c878..4bd415685110b 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -2015,7 +2015,7 @@ func getUintFromNode(ctx sessionctx.Context, n ast.Node, mustInt64orUint64 bool)
 			return 0, false, true
 		}
 		if mustInt64orUint64 {
-			if expected := checkParamTypeInt64orUint64(v); !expected {
+			if expected, _ := CheckParamTypeInt64orUint64(v); !expected {
 				return 0, false, false
 			}
 		}
@@ -2052,19 +2052,19 @@ func getUintFromNode(ctx sessionctx.Context, n ast.Node, mustInt64orUint64 bool)
 	return 0, false, false
 }
 
-// check param type for plan cache limit, only allow int64 and uint64 now
+// CheckParamTypeInt64orUint64 check param type for plan cache limit, only allow int64 and uint64 now
 // eg: set @a = 1;
-func checkParamTypeInt64orUint64(param *driver.ParamMarkerExpr) bool {
+func CheckParamTypeInt64orUint64(param *driver.ParamMarkerExpr) (bool, uint64) {
 	val := param.GetValue()
 	switch v := val.(type) {
 	case int64:
 		if v >= 0 {
-			return true
+			return true, uint64(v)
 		}
 	case uint64:
-		return true
+		return true, v
 	}
-	return false
+	return false, 0
 }
 
 func extractLimitCountOffset(ctx sessionctx.Context, limit *ast.Limit) (count uint64,
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 7c72762394dc4..c112d1fb7d956 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -472,7 +472,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 	case *ast.Limit:
 		if node.Count != nil {
 			if count, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
-				typeExpected, val := checkLimitParamType(count)
+				typeExpected, val := CheckParamTypeInt64orUint64(count)
 				if typeExpected {
 					if val > 10000 {
 						checker.cacheable = false
@@ -488,7 +488,7 @@ func (checker *limitExtractor) Enter(in ast.Node) (out ast.Node, skipChildren bo
 		}
 		if node.Offset != nil {
 			if offset, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
-				typeExpected, val := checkLimitParamType(offset)
+				typeExpected, val := CheckParamTypeInt64orUint64(offset)
 				if typeExpected {
 					checker.offsetAndCount = append(checker.offsetAndCount, val)
 				} else {
@@ -524,14 +524,3 @@ func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]uint64, erro
 	}
 	return checker.offsetAndCount, nil
 }
-
-func checkLimitParamType(node *driver.ParamMarkerExpr) (bool, uint64) {
-	val := node.GetValue()
-	switch v := val.(type) {
-	case int64:
-		return true, uint64(v)
-	case uint64:
-		return true, v
-	}
-	return false, 0
-}

From de1d441ac3276a48c2e2adf90ab31845b162adea Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Thu, 12 Jan 2023 16:15:44 +0800
Subject: [PATCH 26/31] Update plan_cache_test.go

---
 planner/core/plan_cache_test.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index ce85b7454d577..5fbf1ca458d08 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -479,7 +479,6 @@ func TestPlanCacheWithLimit(t *testing.T) {
 
 	for _, testCase := range testCases {
 		tk.MustExec(testCase.sql)
-		tk.MustExec("set @a = 1")
 		var using []string
 		for i, p := range testCase.params {
 			tk.MustExec(fmt.Sprintf("set @a%d = %d", i, p))
@@ -490,7 +489,7 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("1"))
 
-		tk.MustExec("set @a0 = 10086")
+		tk.MustExec("set @a0 = 6")
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
 	}

From 4c05386704f9cfd4b1f97243ca9468f7f137e319 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Fri, 13 Jan 2023 09:47:41 +0800
Subject: [PATCH 27/31] move limit params

---
 executor/prepared.go                  |  6 +---
 planner/core/plan_cache.go            | 30 +++++++++---------
 planner/core/plan_cache_lru.go        | 45 ++++++++++++++++++++++-----
 planner/core/plan_cache_lru_test.go   | 32 +++++++++----------
 planner/core/plan_cache_utils.go      | 29 ++++++++---------
 planner/core/plan_cache_utils_test.go |  8 ++---
 server/driver_tidb.go                 |  6 +---
 session/session.go                    |  7 +----
 sessionctx/context.go                 |  4 +--
 9 files changed, 91 insertions(+), 76 deletions(-)

diff --git a/executor/prepared.go b/executor/prepared.go
index a93adcb3d6368..6a5025e0d539b 100644
--- a/executor/prepared.go
+++ b/executor/prepared.go
@@ -204,12 +204,8 @@ func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
 	delete(vars.PreparedStmtNameToID, e.Name)
 	if e.ctx.GetSessionVars().EnablePreparedPlanCache {
 		bindSQL, _ := plannercore.GetBindSQL4PlanCache(e.ctx, preparedObj)
-		limitCountAndOffset, paramErr := plannercore.ExtractLimitFromAst(prepared.Stmt, nil)
-		if paramErr != nil {
-			return paramErr
-		}
 		cacheKey, err := plannercore.NewPlanCacheKey(vars, preparedObj.StmtText, preparedObj.StmtDB, prepared.SchemaVersion,
-			0, bindSQL, limitCountAndOffset)
+			0, bindSQL)
 		if err != nil {
 			return err
 		}
diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index 5a9f6f92905e1..c9016e85955c5 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -152,12 +152,8 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			// up-to-date schema version which can lead plan cache miss and thus, the plan will be rebuilt.
 			latestSchemaVersion = domain.GetDomain(sctx).InfoSchema().SchemaMetaVersion()
 		}
-		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)
-		if paramErr != nil {
-			return nil, nil, paramErr
-		}
 		if cacheKey, err = NewPlanCacheKey(sctx.GetSessionVars(), stmt.StmtText,
-			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitCountAndOffset); err != nil {
+			stmt.StmtDB, stmtAst.SchemaVersion, latestSchemaVersion, bindSQL); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -171,8 +167,12 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 	}
 
 	if stmtCtx.UseCache { // for non-point plans
+		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)
+		if paramErr != nil {
+			return nil, nil, paramErr
+		}
 		if plan, names, ok, err := getCachedPlan(sctx, isNonPrepared, cacheKey, bindSQL, is, stmt,
-			paramTypes); err != nil || ok {
+			paramTypes, limitCountAndOffset); err != nil || ok {
 			return plan, names, err
 		}
 	}
@@ -225,12 +225,12 @@ func getCachedPointPlan(stmt *ast.Prepared, sessVars *variable.SessionVars, stmt
 }
 
 func getCachedPlan(sctx sessionctx.Context, isNonPrepared bool, cacheKey kvcache.Key, bindSQL string,
-	is infoschema.InfoSchema, stmt *PlanCacheStmt, paramTypes []*types.FieldType) (Plan,
+	is infoschema.InfoSchema, stmt *PlanCacheStmt, paramTypes []*types.FieldType, limitParams []uint64) (Plan,
 	[]*types.FieldName, bool, error) {
 	sessVars := sctx.GetSessionVars()
 	stmtCtx := sessVars.StmtCtx
 
-	candidate, exist := sctx.GetPlanCache(isNonPrepared).Get(cacheKey, paramTypes)
+	candidate, exist := sctx.GetPlanCache(isNonPrepared).Get(cacheKey, paramTypes, limitParams)
 	if !exist {
 		return nil, nil, false, nil
 	}
@@ -294,21 +294,21 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 		// rebuild key to exclude kv.TiFlash when stmt is not read only
 		if _, isolationReadContainTiFlash := sessVars.IsolationReadEngines[kv.TiFlash]; isolationReadContainTiFlash && !IsReadOnly(stmtAst.Stmt, sessVars) {
 			delete(sessVars.IsolationReadEngines, kv.TiFlash)
-			limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)
-			if paramErr != nil {
-				return nil, nil, paramErr
-			}
 			if cacheKey, err = NewPlanCacheKey(sessVars, stmt.StmtText, stmt.StmtDB,
-				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL, limitCountAndOffset); err != nil {
+				stmtAst.SchemaVersion, latestSchemaVersion, bindSQL); err != nil {
 				return nil, nil, err
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
 		}
-		cached := NewPlanCacheValue(p, names, stmtCtx.TblInfo2UnionScan, paramTypes)
+		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)
+		if paramErr != nil {
+			return nil, nil, paramErr
+		}
+		cached := NewPlanCacheValue(p, names, stmtCtx.TblInfo2UnionScan, paramTypes, limitCountAndOffset)
 		stmt.NormalizedPlan, stmt.PlanDigest = NormalizePlan(p)
 		stmtCtx.SetPlan(p)
 		stmtCtx.SetPlanDigest(stmt.NormalizedPlan, stmt.PlanDigest)
-		sctx.GetPlanCache(isNonPrepared).Put(cacheKey, cached, paramTypes)
+		sctx.GetPlanCache(isNonPrepared).Put(cacheKey, cached, paramTypes, limitCountAndOffset)
 	}
 	sessVars.FoundInPlanCache = false
 	return p, names, err
diff --git a/planner/core/plan_cache_lru.go b/planner/core/plan_cache_lru.go
index 413dd37e8f5a2..30e40a6dd20c3 100644
--- a/planner/core/plan_cache_lru.go
+++ b/planner/core/plan_cache_lru.go
@@ -53,7 +53,7 @@ type LRUPlanCache struct {
 	lock sync.Mutex
 
 	// pickFromBucket get one element from bucket. The LRUPlanCache can not work if it is nil
-	pickFromBucket func(map[*list.Element]struct{}, []*types.FieldType) (*list.Element, bool)
+	pickFromBucket func(map[*list.Element]struct{}, []*types.FieldType, []uint64) (*list.Element, bool)
 	// onEvict will be called if any eviction happened, only for test use now
 	onEvict func(kvcache.Key, kvcache.Value)
 
@@ -68,7 +68,7 @@ type LRUPlanCache struct {
 // NewLRUPlanCache creates a PCLRUCache object, whose capacity is "capacity".
 // NOTE: "capacity" should be a positive value.
 func NewLRUPlanCache(capacity uint, guard float64, quota uint64,
-	pickFromBucket func(map[*list.Element]struct{}, []*types.FieldType) (*list.Element, bool), sctx sessionctx.Context) *LRUPlanCache {
+	pickFromBucket func(map[*list.Element]struct{}, []*types.FieldType, []uint64) (*list.Element, bool), sctx sessionctx.Context) *LRUPlanCache {
 	if capacity < 1 {
 		capacity = 100
 		logutil.BgLogger().Info("capacity of LRU cache is less than 1, will use default value(100) init cache")
@@ -94,13 +94,13 @@ func strHashKey(key kvcache.Key, deepCopy bool) string {
 }
 
 // Get tries to find the corresponding value according to the given key.
-func (l *LRUPlanCache) Get(key kvcache.Key, paramTypes []*types.FieldType) (value kvcache.Value, ok bool) {
+func (l *LRUPlanCache) Get(key kvcache.Key, paramTypes []*types.FieldType, limitParams []uint64) (value kvcache.Value, ok bool) {
 	l.lock.Lock()
 	defer l.lock.Unlock()
 
 	bucket, bucketExist := l.buckets[strHashKey(key, false)]
 	if bucketExist {
-		if element, exist := l.pickFromBucket(bucket, paramTypes); exist {
+		if element, exist := l.pickFromBucket(bucket, paramTypes, limitParams); exist {
 			l.lruList.MoveToFront(element)
 			return element.Value.(*planCacheEntry).PlanValue, true
 		}
@@ -109,14 +109,14 @@ func (l *LRUPlanCache) Get(key kvcache.Key, paramTypes []*types.FieldType) (valu
 }
 
 // Put puts the (key, value) pair into the LRU Cache.
-func (l *LRUPlanCache) Put(key kvcache.Key, value kvcache.Value, paramTypes []*types.FieldType) {
+func (l *LRUPlanCache) Put(key kvcache.Key, value kvcache.Value, paramTypes []*types.FieldType, limitParams []uint64) {
 	l.lock.Lock()
 	defer l.lock.Unlock()
 
 	hash := strHashKey(key, true)
 	bucket, bucketExist := l.buckets[hash]
 	if bucketExist {
-		if element, exist := l.pickFromBucket(bucket, paramTypes); exist {
+		if element, exist := l.pickFromBucket(bucket, paramTypes, limitParams); exist {
 			l.updateInstanceMetric(&planCacheEntry{PlanKey: key, PlanValue: value}, element.Value.(*planCacheEntry))
 			element.Value.(*planCacheEntry).PlanValue = value
 			l.lruList.MoveToFront(element)
@@ -252,7 +252,7 @@ func (l *LRUPlanCache) memoryControl() {
 }
 
 // PickPlanFromBucket pick one plan from bucket
-func PickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.FieldType) (*list.Element, bool) {
+func oldPickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.FieldType) (*list.Element, bool) {
 	for k := range bucket {
 		plan := k.Value.(*planCacheEntry).PlanValue.(*PlanCacheValue)
 		if plan.ParamTypes.CheckTypesCompatibility4PC(paramTypes) {
@@ -262,6 +262,37 @@ func PickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.F
 	return nil, false
 }
 
+// PickPlanFromBucket pick one plan from bucket
+func PickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.FieldType, limitParams []uint64) (*list.Element, bool) {
+	for k := range bucket {
+		plan := k.Value.(*planCacheEntry).PlanValue.(*PlanCacheValue)
+		ok1 := plan.ParamTypes.CheckTypesCompatibility4PC(paramTypes)
+		if !ok1 {
+			continue
+		}
+		ok2 := checkUint64SliceIfEqual(plan.limitOffsetAndCount, limitParams)
+		if ok2 {
+			return k, true
+		}
+	}
+	return nil, false
+}
+
+func checkUint64SliceIfEqual(a, b []uint64) bool {
+	if (a == nil && b != nil) || (a != nil && b == nil) {
+		return false
+	}
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
 // updateInstanceMetric update the memory usage and plan num for show in grafana
 func (l *LRUPlanCache) updateInstanceMetric(in, out *planCacheEntry) {
 	updateInstancePlanNum(in, out)
diff --git a/planner/core/plan_cache_lru_test.go b/planner/core/plan_cache_lru_test.go
index 74b6b2c92c3bb..d67b0db7171f4 100644
--- a/planner/core/plan_cache_lru_test.go
+++ b/planner/core/plan_cache_lru_test.go
@@ -72,7 +72,7 @@ func TestLRUPCPut(t *testing.T) {
 		vals[i] = &PlanCacheValue{
 			ParamTypes: pTypes[i],
 		}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, lru.size, lru.capacity)
 	require.Equal(t, uint(3), lru.size)
@@ -103,7 +103,7 @@ func TestLRUPCPut(t *testing.T) {
 
 		bucket, exist := lru.buckets[string(hack.String(keys[i].Hash()))]
 		require.True(t, exist)
-		element, exist := lru.pickFromBucket(bucket, pTypes[i])
+		element, exist := lru.pickFromBucket(bucket, pTypes[i], []uint64{})
 		require.NotNil(t, element)
 		require.True(t, exist)
 		require.Equal(t, root, element)
@@ -135,18 +135,18 @@ func TestLRUPCGet(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i%4), 10)}
 		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 
 	// test for non-existent elements
 	for i := 0; i < 2; i++ {
-		value, exists := lru.Get(keys[i], pTypes[i])
+		value, exists := lru.Get(keys[i], pTypes[i], []uint64{})
 		require.False(t, exists)
 		require.Nil(t, value)
 	}
 
 	for i := 2; i < 5; i++ {
-		value, exists := lru.Get(keys[i], pTypes[i])
+		value, exists := lru.Get(keys[i], pTypes[i], []uint64{})
 		require.True(t, exists)
 		require.NotNil(t, value)
 		require.Equal(t, vals[i], value)
@@ -178,20 +178,20 @@ func TestLRUPCDelete(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i), 10)}
 		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, 3, int(lru.size))
 
 	lru.Delete(keys[1])
-	value, exists := lru.Get(keys[1], pTypes[1])
+	value, exists := lru.Get(keys[1], pTypes[1], []uint64{})
 	require.False(t, exists)
 	require.Nil(t, value)
 	require.Equal(t, 2, int(lru.size))
 
-	_, exists = lru.Get(keys[0], pTypes[0])
+	_, exists = lru.Get(keys[0], pTypes[0], []uint64{})
 	require.True(t, exists)
 
-	_, exists = lru.Get(keys[2], pTypes[2])
+	_, exists = lru.Get(keys[2], pTypes[2], []uint64{})
 	require.True(t, exists)
 }
 
@@ -207,14 +207,14 @@ func TestLRUPCDeleteAll(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i), 10)}
 		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, 3, int(lru.size))
 
 	lru.DeleteAll()
 
 	for i := 0; i < 3; i++ {
-		value, exists := lru.Get(keys[i], pTypes[i])
+		value, exists := lru.Get(keys[i], pTypes[i], []uint64{})
 		require.False(t, exists)
 		require.Nil(t, value)
 		require.Equal(t, 0, int(lru.size))
@@ -242,7 +242,7 @@ func TestLRUPCSetCapacity(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(1), 10)}
 		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, lru.size, lru.capacity)
 	require.Equal(t, uint(5), lru.size)
@@ -292,7 +292,7 @@ func TestIssue37914(t *testing.T) {
 	val := &PlanCacheValue{ParamTypes: pTypes}
 
 	require.NotPanics(t, func() {
-		lru.Put(key, val, pTypes)
+		lru.Put(key, val, pTypes, []uint64{})
 	})
 }
 
@@ -313,7 +313,7 @@ func TestIssue38244(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i), 10)}
 		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i])
+		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, lru.size, lru.capacity)
 	require.Equal(t, uint(3), lru.size)
@@ -334,7 +334,7 @@ func TestLRUPlanCacheMemoryUsage(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		k := randomPlanCacheKey()
 		v := randomPlanCacheValue(pTypes)
-		lru.Put(k, v, pTypes)
+		lru.Put(k, v, pTypes, []uint64{})
 		res += k.MemoryUsage() + v.MemoryUsage()
 		require.Equal(t, lru.MemoryUsage(), res)
 	}
@@ -342,7 +342,7 @@ func TestLRUPlanCacheMemoryUsage(t *testing.T) {
 	p := &PhysicalTableScan{}
 	k := &planCacheKey{database: "3"}
 	v := &PlanCacheValue{Plan: p}
-	lru.Put(k, v, pTypes)
+	lru.Put(k, v, pTypes, []uint64{})
 	res += k.MemoryUsage() + v.MemoryUsage()
 	for kk, vv := range evict {
 		res -= kk.(*planCacheKey).MemoryUsage() + vv.(*PlanCacheValue).MemoryUsage()
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index c112d1fb7d956..90e2f97ef0642 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -194,7 +194,6 @@ type planCacheKey struct {
 	inRestrictedSQL          bool
 	restrictedReadOnly       bool
 	TiDBSuperReadOnly        bool
-	limitOffsetAndCount      []uint64
 
 	memoryUsage int64 // Do not include in hash
 	hash        []byte
@@ -231,9 +230,6 @@ func (key *planCacheKey) Hash() []byte {
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.inRestrictedSQL))...)
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.restrictedReadOnly))...)
 		key.hash = append(key.hash, hack.Slice(strconv.FormatBool(key.TiDBSuperReadOnly))...)
-		for _, l := range key.limitOffsetAndCount {
-			key.hash = codec.EncodeUint(key.hash, l)
-		}
 	}
 	return key.hash
 }
@@ -275,7 +271,7 @@ func SetPstmtIDSchemaVersion(key kvcache.Key, stmtText string, schemaVersion int
 // Note: lastUpdatedSchemaVersion will only be set in the case of rc or for update read in order to
 // differentiate the cache key. In other cases, it will be 0.
 func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string, schemaVersion int64,
-	lastUpdatedSchemaVersion int64, bindSQL string, offsetAndCount []uint64) (kvcache.Key, error) {
+	lastUpdatedSchemaVersion int64, bindSQL string) (kvcache.Key, error) {
 	if stmtText == "" {
 		return nil, errors.New("no statement text")
 	}
@@ -303,7 +299,6 @@ func NewPlanCacheKey(sessionVars *variable.SessionVars, stmtText, stmtDB string,
 		inRestrictedSQL:          sessionVars.InRestrictedSQL,
 		restrictedReadOnly:       variable.RestrictedReadOnly.Load(),
 		TiDBSuperReadOnly:        variable.VarTiDBSuperReadOnly.Load(),
-		limitOffsetAndCount:      offsetAndCount,
 	}
 	for k, v := range sessionVars.IsolationReadEngines {
 		key.isolationReadEngines[k] = v
@@ -343,11 +338,12 @@ func (s FieldSlice) CheckTypesCompatibility4PC(tps []*types.FieldType) bool {
 
 // PlanCacheValue stores the cached Statement and StmtNode.
 type PlanCacheValue struct {
-	Plan              Plan
-	OutPutNames       []*types.FieldName
-	TblInfo2UnionScan map[*model.TableInfo]bool
-	ParamTypes        FieldSlice
-	memoryUsage       int64
+	Plan                Plan
+	OutPutNames         []*types.FieldName
+	TblInfo2UnionScan   map[*model.TableInfo]bool
+	ParamTypes          FieldSlice
+	memoryUsage         int64
+	limitOffsetAndCount []uint64
 }
 
 func (v *PlanCacheValue) varTypesUnchanged(txtVarTps []*types.FieldType) bool {
@@ -395,7 +391,7 @@ func (v *PlanCacheValue) MemoryUsage() (sum int64) {
 
 // NewPlanCacheValue creates a SQLCacheValue.
 func NewPlanCacheValue(plan Plan, names []*types.FieldName, srcMap map[*model.TableInfo]bool,
-	paramTypes []*types.FieldType) *PlanCacheValue {
+	paramTypes []*types.FieldType, limitParams []uint64) *PlanCacheValue {
 	dstMap := make(map[*model.TableInfo]bool)
 	for k, v := range srcMap {
 		dstMap[k] = v
@@ -405,10 +401,11 @@ func NewPlanCacheValue(plan Plan, names []*types.FieldName, srcMap map[*model.Ta
 		userParamTypes[i] = tp.Clone()
 	}
 	return &PlanCacheValue{
-		Plan:              plan,
-		OutPutNames:       names,
-		TblInfo2UnionScan: dstMap,
-		ParamTypes:        userParamTypes,
+		Plan:                plan,
+		OutPutNames:         names,
+		TblInfo2UnionScan:   dstMap,
+		ParamTypes:          userParamTypes,
+		limitOffsetAndCount: limitParams,
 	}
 }
 
diff --git a/planner/core/plan_cache_utils_test.go b/planner/core/plan_cache_utils_test.go
index b1ac33f56e6ba..6f0938e447263 100644
--- a/planner/core/plan_cache_utils_test.go
+++ b/planner/core/plan_cache_utils_test.go
@@ -32,20 +32,20 @@ func TestCacheKey(t *testing.T) {
 	ctx.GetSessionVars().InRestrictedSQL = false
 	variable.RestrictedReadOnly.Store(false)
 	variable.VarTiDBSuperReadOnly.Store(false)
-	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "", []uint64{})
+	key, err := NewPlanCacheKey(ctx.GetSessionVars(), "", "test", 1, 1, "")
 	if err.Error() != "no statement text" {
 		t.Fail() // no statement text
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "", []uint64{})
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "", 1, 1, "")
 	if err != nil {
 		t.Fail() // schema can be nil
 	}
 	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1,
-		"select /*+ ignore_plan_cache() */ * from t", []uint64{})
+		"select /*+ ignore_plan_cache() */ * from t")
 	if err != nil {
 		t.Fail()
 	}
-	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "", []uint64{})
+	key, err = NewPlanCacheKey(ctx.GetSessionVars(), "select 1", "test", 1, 1, "")
 	if err != nil {
 		t.Fail()
 	}
diff --git a/server/driver_tidb.go b/server/driver_tidb.go
index a822fa014ef2c..7b25a998d618b 100644
--- a/server/driver_tidb.go
+++ b/server/driver_tidb.go
@@ -171,12 +171,8 @@ func (ts *TiDBStatement) Close() error {
 				return errors.Errorf("invalid PlanCacheStmt type")
 			}
 			bindSQL, _ := core.GetBindSQL4PlanCache(ts.ctx, preparedObj)
-			limitCountAndOffset, paramErr := core.ExtractLimitFromAst(preparedObj.PreparedAst.Stmt, nil)
-			if paramErr != nil {
-				return paramErr
-			}
 			cacheKey, err := core.NewPlanCacheKey(ts.ctx.GetSessionVars(), preparedObj.StmtText, preparedObj.StmtDB,
-				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL, limitCountAndOffset)
+				preparedObj.PreparedAst.SchemaVersion, 0, bindSQL)
 			if err != nil {
 				return err
 			}
diff --git a/session/session.go b/session/session.go
index a87ea922145b9..f6dbed8dbadc2 100644
--- a/session/session.go
+++ b/session/session.go
@@ -384,13 +384,8 @@ func (s *session) cleanRetryInfo() {
 				preparedAst = preparedObj.PreparedAst
 				stmtText, stmtDB = preparedObj.StmtText, preparedObj.StmtDB
 				bindSQL, _ := plannercore.GetBindSQL4PlanCache(s, preparedObj)
-				limitCountAndOffset, paramErr := plannercore.ExtractLimitFromAst(preparedAst.Stmt, nil)
-				if paramErr != nil {
-					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(paramErr))
-					return
-				}
 				cacheKey, err = plannercore.NewPlanCacheKey(s.sessionVars, stmtText, stmtDB, preparedAst.SchemaVersion,
-					0, bindSQL, limitCountAndOffset)
+					0, bindSQL)
 				if err != nil {
 					logutil.Logger(s.currentCtx).Warn("clean cached plan failed", zap.Error(err))
 					return
diff --git a/sessionctx/context.go b/sessionctx/context.go
index 4cc201206df07..f0e2cd5b9c3b2 100644
--- a/sessionctx/context.go
+++ b/sessionctx/context.go
@@ -54,8 +54,8 @@ type SessionStatesHandler interface {
 
 // PlanCache is an interface for prepare and non-prepared plan cache
 type PlanCache interface {
-	Get(key kvcache.Key, paramTypes []*types.FieldType) (value kvcache.Value, ok bool)
-	Put(key kvcache.Key, value kvcache.Value, paramTypes []*types.FieldType)
+	Get(key kvcache.Key, paramTypes []*types.FieldType, limitParams []uint64) (value kvcache.Value, ok bool)
+	Put(key kvcache.Key, value kvcache.Value, paramTypes []*types.FieldType, limitParams []uint64)
 	Delete(key kvcache.Key)
 	DeleteAll()
 	Size() int

From 67f83dcc0ead52e6f6c616cb151fed1c21e19b13 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Fri, 13 Jan 2023 10:20:07 +0800
Subject: [PATCH 28/31] Update plan_cache_lru_test.go

---
 planner/core/plan_cache_lru_test.go | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/planner/core/plan_cache_lru_test.go b/planner/core/plan_cache_lru_test.go
index d67b0db7171f4..a9fa748a7d537 100644
--- a/planner/core/plan_cache_lru_test.go
+++ b/planner/core/plan_cache_lru_test.go
@@ -131,22 +131,25 @@ func TestLRUPCGet(t *testing.T) {
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeLong)},
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeInt24)},
 	}
+	limitParams := [][]uint64{
+		{1}, {2}, {3}, {4}, {5},
+	}
 	// 5 bucket
 	for i := 0; i < 5; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i%4), 10)}
-		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
-		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
+		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i], limitOffsetAndCount: limitParams[i]}
+		lru.Put(keys[i], vals[i], pTypes[i], limitParams[i])
 	}
 
 	// test for non-existent elements
 	for i := 0; i < 2; i++ {
-		value, exists := lru.Get(keys[i], pTypes[i], []uint64{})
+		value, exists := lru.Get(keys[i], pTypes[i], limitParams[i])
 		require.False(t, exists)
 		require.Nil(t, value)
 	}
 
 	for i := 2; i < 5; i++ {
-		value, exists := lru.Get(keys[i], pTypes[i], []uint64{})
+		value, exists := lru.Get(keys[i], pTypes[i], limitParams[i])
 		require.True(t, exists)
 		require.NotNil(t, value)
 		require.Equal(t, vals[i], value)

From a9ba0e6608546004c6c77b78fceec2930927fd95 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Fri, 13 Jan 2023 10:48:18 +0800
Subject: [PATCH 29/31] fix ut

---
 planner/core/plan_cache_lru_test.go | 24 +++++++++++++++++-------
 1 file changed, 17 insertions(+), 7 deletions(-)

diff --git a/planner/core/plan_cache_lru_test.go b/planner/core/plan_cache_lru_test.go
index a9fa748a7d537..f51480401ce62 100644
--- a/planner/core/plan_cache_lru_test.go
+++ b/planner/core/plan_cache_lru_test.go
@@ -65,14 +65,18 @@ func TestLRUPCPut(t *testing.T) {
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeLong)},
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeInt24)},
 	}
+	limitParams := [][]uint64{
+		{1}, {2}, {3}, {4}, {5},
+	}
 
 	// one key corresponding to multi values
 	for i := 0; i < 5; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(1), 10)}
 		vals[i] = &PlanCacheValue{
-			ParamTypes: pTypes[i],
+			ParamTypes:          pTypes[i],
+			limitOffsetAndCount: limitParams[i],
 		}
-		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
+		lru.Put(keys[i], vals[i], pTypes[i], limitParams[i])
 	}
 	require.Equal(t, lru.size, lru.capacity)
 	require.Equal(t, uint(3), lru.size)
@@ -103,7 +107,7 @@ func TestLRUPCPut(t *testing.T) {
 
 		bucket, exist := lru.buckets[string(hack.String(keys[i].Hash()))]
 		require.True(t, exist)
-		element, exist := lru.pickFromBucket(bucket, pTypes[i], []uint64{})
+		element, exist := lru.pickFromBucket(bucket, pTypes[i], limitParams[i])
 		require.NotNil(t, element)
 		require.True(t, exist)
 		require.Equal(t, root, element)
@@ -178,23 +182,29 @@ func TestLRUPCDelete(t *testing.T) {
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeEnum)},
 		{types.NewFieldType(mysql.TypeFloat), types.NewFieldType(mysql.TypeDate)},
 	}
+	limitParams := [][]uint64{
+		{1}, {2}, {3},
+	}
 	for i := 0; i < 3; i++ {
 		keys[i] = &planCacheKey{database: strconv.FormatInt(int64(i), 10)}
-		vals[i] = &PlanCacheValue{ParamTypes: pTypes[i]}
+		vals[i] = &PlanCacheValue{
+			ParamTypes:          pTypes[i],
+			limitOffsetAndCount: limitParams[i],
+		}
 		lru.Put(keys[i], vals[i], pTypes[i], []uint64{})
 	}
 	require.Equal(t, 3, int(lru.size))
 
 	lru.Delete(keys[1])
-	value, exists := lru.Get(keys[1], pTypes[1], []uint64{})
+	value, exists := lru.Get(keys[1], pTypes[1], limitParams[1])
 	require.False(t, exists)
 	require.Nil(t, value)
 	require.Equal(t, 2, int(lru.size))
 
-	_, exists = lru.Get(keys[0], pTypes[0], []uint64{})
+	_, exists = lru.Get(keys[0], pTypes[0], limitParams[0])
 	require.True(t, exists)
 
-	_, exists = lru.Get(keys[2], pTypes[2], []uint64{})
+	_, exists = lru.Get(keys[2], pTypes[2], limitParams[2])
 	require.True(t, exists)
 }
 

From e1b33a4f26c22af5a997a192ab3bb6f379312d11 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Fri, 13 Jan 2023 11:12:27 +0800
Subject: [PATCH 30/31] Update plan_cache_lru.go

---
 planner/core/plan_cache_lru.go | 11 -----------
 1 file changed, 11 deletions(-)

diff --git a/planner/core/plan_cache_lru.go b/planner/core/plan_cache_lru.go
index 30e40a6dd20c3..20fa4c3f5c85c 100644
--- a/planner/core/plan_cache_lru.go
+++ b/planner/core/plan_cache_lru.go
@@ -251,17 +251,6 @@ func (l *LRUPlanCache) memoryControl() {
 	}
 }
 
-// PickPlanFromBucket pick one plan from bucket
-func oldPickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.FieldType) (*list.Element, bool) {
-	for k := range bucket {
-		plan := k.Value.(*planCacheEntry).PlanValue.(*PlanCacheValue)
-		if plan.ParamTypes.CheckTypesCompatibility4PC(paramTypes) {
-			return k, true
-		}
-	}
-	return nil, false
-}
-
 // PickPlanFromBucket pick one plan from bucket
 func PickPlanFromBucket(bucket map[*list.Element]struct{}, paramTypes []*types.FieldType, limitParams []uint64) (*list.Element, bool) {
 	for k := range bucket {

From b877d21429f1156dea0728e02fdba63fc636b9c4 Mon Sep 17 00:00:00 2001
From: fzzf678 <108643977+fzzf678@users.noreply.github.com>
Date: Mon, 16 Jan 2023 14:48:25 +0800
Subject: [PATCH 31/31] fix

---
 planner/core/plan_cache.go             | 21 ++++++++-------------
 planner/core/plan_cache_test.go        | 16 ++++++++++------
 planner/core/plan_cache_utils.go       | 16 +++++++++-------
 planner/core/plan_cacheable_checker.go | 16 ++++++++++++++++
 4 files changed, 43 insertions(+), 26 deletions(-)

diff --git a/planner/core/plan_cache.go b/planner/core/plan_cache.go
index c9016e85955c5..70ae7d3481616 100644
--- a/planner/core/plan_cache.go
+++ b/planner/core/plan_cache.go
@@ -165,19 +165,18 @@ func GetPlanFromSessionPlanCache(ctx context.Context, sctx sessionctx.Context,
 			return plan, names, err
 		}
 	}
-
+	limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)
+	if paramErr != nil {
+		return nil, nil, paramErr
+	}
 	if stmtCtx.UseCache { // for non-point plans
-		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, sctx)
-		if paramErr != nil {
-			return nil, nil, paramErr
-		}
 		if plan, names, ok, err := getCachedPlan(sctx, isNonPrepared, cacheKey, bindSQL, is, stmt,
 			paramTypes, limitCountAndOffset); err != nil || ok {
 			return plan, names, err
 		}
 	}
 
-	return generateNewPlan(ctx, sctx, isNonPrepared, is, stmt, cacheKey, latestSchemaVersion, paramNum, paramTypes, bindSQL)
+	return generateNewPlan(ctx, sctx, isNonPrepared, is, stmt, cacheKey, latestSchemaVersion, paramNum, paramTypes, bindSQL, limitCountAndOffset)
 }
 
 // parseParamTypes get parameters' types in PREPARE statement
@@ -269,7 +268,7 @@ func getCachedPlan(sctx sessionctx.Context, isNonPrepared bool, cacheKey kvcache
 // generateNewPlan call the optimizer to generate a new plan for current statement
 // and try to add it to cache
 func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared bool, is infoschema.InfoSchema, stmt *PlanCacheStmt, cacheKey kvcache.Key, latestSchemaVersion int64, paramNum int,
-	paramTypes []*types.FieldType, bindSQL string) (Plan, []*types.FieldName, error) {
+	paramTypes []*types.FieldType, bindSQL string, limitParams []uint64) (Plan, []*types.FieldName, error) {
 	stmtAst := stmt.PreparedAst
 	sessVars := sctx.GetSessionVars()
 	stmtCtx := sessVars.StmtCtx
@@ -300,15 +299,11 @@ func generateNewPlan(ctx context.Context, sctx sessionctx.Context, isNonPrepared
 			}
 			sessVars.IsolationReadEngines[kv.TiFlash] = struct{}{}
 		}
-		limitCountAndOffset, paramErr := ExtractLimitFromAst(stmt.PreparedAst.Stmt, nil)
-		if paramErr != nil {
-			return nil, nil, paramErr
-		}
-		cached := NewPlanCacheValue(p, names, stmtCtx.TblInfo2UnionScan, paramTypes, limitCountAndOffset)
+		cached := NewPlanCacheValue(p, names, stmtCtx.TblInfo2UnionScan, paramTypes, limitParams)
 		stmt.NormalizedPlan, stmt.PlanDigest = NormalizePlan(p)
 		stmtCtx.SetPlan(p)
 		stmtCtx.SetPlanDigest(stmt.NormalizedPlan, stmt.PlanDigest)
-		sctx.GetPlanCache(isNonPrepared).Put(cacheKey, cached, paramTypes, limitCountAndOffset)
+		sctx.GetPlanCache(isNonPrepared).Put(cacheKey, cached, paramTypes, limitParams)
 	}
 	sessVars.FoundInPlanCache = false
 	return p, names, err
diff --git a/planner/core/plan_cache_test.go b/planner/core/plan_cache_test.go
index 5fbf1ca458d08..8acc28b7b0062 100644
--- a/planner/core/plan_cache_test.go
+++ b/planner/core/plan_cache_test.go
@@ -462,7 +462,7 @@ func TestPlanCacheWithLimit(t *testing.T) {
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists t")
-	tk.MustExec("create table t(a int, b int, key(a))")
+	tk.MustExec("create table t(a int primary key, b int)")
 
 	testCases := []struct {
 		sql    string
@@ -474,10 +474,12 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		{"prepare stmt from 'insert into t select * from t order by a desc limit ?'", []int{1}},
 		{"prepare stmt from 'insert into t select * from t order by a desc limit ?, ?'", []int{1, 2}},
 		{"prepare stmt from 'update t set a = 1 limit ?'", []int{1}},
-		{" prepare stmt from '(select * from t order by a limit ?) union (select * from t order by a desc limit ?)';", []int{1, 2}},
+		{"prepare stmt from '(select * from t order by a limit ?) union (select * from t order by a desc limit ?)'", []int{1, 2}},
+		{"prepare stmt from 'select * from t where a = ? limit ?, ?'", []int{1, 1, 1}},
+		{"prepare stmt from 'select * from t where a in (?, ?) limit ?, ?'", []int{1, 2, 1, 1}},
 	}
 
-	for _, testCase := range testCases {
+	for idx, testCase := range testCases {
 		tk.MustExec(testCase.sql)
 		var using []string
 		for i, p := range testCase.params {
@@ -489,9 +491,11 @@ func TestPlanCacheWithLimit(t *testing.T) {
 		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
 		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("1"))
 
-		tk.MustExec("set @a0 = 6")
-		tk.MustExec("execute stmt using " + strings.Join(using, ", "))
-		tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
+		if idx < 6 {
+			tk.MustExec("set @a0 = 6")
+			tk.MustExec("execute stmt using " + strings.Join(using, ", "))
+			tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
+		}
 	}
 
 	tk.MustExec("prepare stmt from 'select * from t limit ?'")
diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go
index 90e2f97ef0642..082637506e590 100644
--- a/planner/core/plan_cache_utils.go
+++ b/planner/core/plan_cache_utils.go
@@ -338,11 +338,13 @@ func (s FieldSlice) CheckTypesCompatibility4PC(tps []*types.FieldType) bool {
 
 // PlanCacheValue stores the cached Statement and StmtNode.
 type PlanCacheValue struct {
-	Plan                Plan
-	OutPutNames         []*types.FieldName
-	TblInfo2UnionScan   map[*model.TableInfo]bool
-	ParamTypes          FieldSlice
-	memoryUsage         int64
+	Plan              Plan
+	OutPutNames       []*types.FieldName
+	TblInfo2UnionScan map[*model.TableInfo]bool
+	ParamTypes        FieldSlice
+	memoryUsage       int64
+	// limitOffsetAndCount stores all the offset and key parameters extract from limit statement
+	// only used for cache and pick plan with parameters
 	limitOffsetAndCount []uint64
 }
 
@@ -506,7 +508,7 @@ func (checker *limitExtractor) Leave(in ast.Node) (out ast.Node, ok bool) {
 // ExtractLimitFromAst extract limit offset and count from ast for plan cache key encode
 func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]uint64, error) {
 	if node == nil {
-		return []uint64{}, nil
+		return nil, nil
 	}
 	checker := limitExtractor{
 		cacheable:      true,
@@ -514,7 +516,7 @@ func ExtractLimitFromAst(node ast.Node, sctx sessionctx.Context) ([]uint64, erro
 	}
 	node.Accept(&checker)
 	if checker.paramTypeErr != nil {
-		return []uint64{}, checker.paramTypeErr
+		return nil, checker.paramTypeErr
 	}
 	if sctx != nil && !checker.cacheable {
 		sctx.GetSessionVars().StmtCtx.SetSkipPlanCache(errors.New("skip plan-cache: " + checker.unCacheableReason))
diff --git a/planner/core/plan_cacheable_checker.go b/planner/core/plan_cacheable_checker.go
index d6f401a2d2d02..2ff9e51823ee2 100644
--- a/planner/core/plan_cacheable_checker.go
+++ b/planner/core/plan_cacheable_checker.go
@@ -135,6 +135,22 @@ func (checker *cacheableChecker) Enter(in ast.Node) (out ast.Node, skipChildren
 				return in, true
 			}
 		}
+	// todo: these comment is used to add switch in the later pr
+	//case *ast.Limit:
+	//	if node.Count != nil {
+	//		if _, isParamMarker := node.Count.(*driver.ParamMarkerExpr); isParamMarker {
+	//			checker.cacheable = false
+	//			checker.reason = "query has 'limit ?' is un-cacheable"
+	//			return in, true
+	//		}
+	//	}
+	//	if node.Offset != nil {
+	//		if _, isParamMarker := node.Offset.(*driver.ParamMarkerExpr); isParamMarker {
+	//			checker.cacheable = false
+	//			checker.reason = "query has 'limit ?, 10' is un-cacheable"
+	//			return in, true
+	//		}
+	//	}
 	case *ast.FrameBound:
 		if _, ok := node.Expr.(*driver.ParamMarkerExpr); ok {
 			checker.cacheable = false