From e773dfc51873dec2760fe93275f91e05a6961564 Mon Sep 17 00:00:00 2001 From: Ti Chi Robot Date: Mon, 5 Jun 2023 12:07:40 +0800 Subject: [PATCH] *: fix PointGet will return an stale value when `tidb_enable_plan_replayer_capture` is set (#40197) (#44092) close pingcap/tidb#40194 --- executor/adapter.go | 19 ++++++++++++++----- executor/compiler.go | 5 +++++ executor/point_get_test.go | 17 +++++++++++++++++ planner/core/plan_cache_utils.go | 11 +++++++---- session/session.go | 4 ---- sessiontxn/interface.go | 2 ++ sessiontxn/isolation/optimistic.go | 8 +++++++- 7 files changed, 52 insertions(+), 14 deletions(-) diff --git a/executor/adapter.go b/executor/adapter.go index 09d70297ed600..b315a404e98e6 100644 --- a/executor/adapter.go +++ b/executor/adapter.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "fmt" + "math" "runtime/trace" "strconv" "strings" @@ -296,8 +297,12 @@ func (a *ExecStmt) PointGet(ctx context.Context) (*recordSet, error) { } a.Ctx.GetSessionVars().StmtCtx.Priority = kv.PriorityHigh + var pointExecutor *PointGetExecutor + useMaxTS := startTs == math.MaxUint64 + // try to reuse point get executor - if a.PsStmt.Executor != nil { + // We should only use the cached the executor when the startTS is MaxUint64 + if a.PsStmt.Executor != nil && useMaxTS { exec, ok := a.PsStmt.Executor.(*PointGetExecutor) if !ok { logutil.Logger(ctx).Error("invalid executor type, not PointGetExecutor for point get path") @@ -307,17 +312,21 @@ func (a *ExecStmt) PointGet(ctx context.Context) (*recordSet, error) { pointGetPlan := a.PsStmt.PreparedAst.CachedPlan.(*plannercore.PointGetPlan) exec.Init(pointGetPlan) a.PsStmt.Executor = exec + pointExecutor = exec } } - if a.PsStmt.Executor == nil { + + if pointExecutor == nil { b := newExecutorBuilder(a.Ctx, a.InfoSchema, a.Ti) - newExecutor := b.build(a.Plan) + pointExecutor = b.build(a.Plan).(*PointGetExecutor) if b.err != nil { return nil, b.err } - a.PsStmt.Executor = newExecutor + + if useMaxTS { + a.PsStmt.Executor = pointExecutor + } } - pointExecutor := a.PsStmt.Executor.(*PointGetExecutor) if err = pointExecutor.Open(ctx); err != nil { terror.Call(pointExecutor.Close) diff --git a/executor/compiler.go b/executor/compiler.go index 6d62b6e3272cf..a5b18aa2c18f0 100644 --- a/executor/compiler.go +++ b/executor/compiler.go @@ -156,6 +156,11 @@ func (c *Compiler) Compile(ctx context.Context, stmtNode ast.StmtNode) (_ *ExecS } } } + + if err = sessiontxn.OptimizeWithPlanAndThenWarmUp(c.Ctx, stmt.Plan); err != nil { + return nil, err + } + if c.Ctx.GetSessionVars().EnablePlanReplayerCapture && !c.Ctx.GetSessionVars().InRestrictedSQL { checkPlanReplayerCaptureTask(c.Ctx, stmtNode) } diff --git a/executor/point_get_test.go b/executor/point_get_test.go index c615c3a75cb1a..8f13675457481 100644 --- a/executor/point_get_test.go +++ b/executor/point_get_test.go @@ -825,3 +825,20 @@ func TestPointGetIssue25167(t *testing.T) { tk.MustExec("insert into t values (1)") tk.MustQuery("select * from t as of timestamp @a where a = 1").Check(testkit.Rows()) } + +func TestPointGetIssue40194(t *testing.T) { + store := testkit.CreateMockStore(t) + tk := testkit.NewTestKit(t, store) + tk.MustExec("use test") + tk.MustExec("create table t1(id int primary key, v int)") + tk.MustExec("insert into t1 values(1, 10)") + tk.MustExec("prepare s from 'select * from t1 where id=1'") + tk.MustExec("set @@tidb_enable_plan_replayer_capture=1") + tk.MustQuery("execute s").Check(testkit.Rows("1 10")) + tk.MustQuery("execute s").Check(testkit.Rows("1 10")) + + tk2 := testkit.NewTestKit(t, store) + tk2.MustExec("use test") + tk2.MustExec("update t1 set v=v+1") + tk.MustQuery("execute s").Check(testkit.Rows("1 11")) +} diff --git a/planner/core/plan_cache_utils.go b/planner/core/plan_cache_utils.go index 4229e2b134f06..5431f7ef71c27 100644 --- a/planner/core/plan_cache_utils.go +++ b/planner/core/plan_cache_utils.go @@ -414,10 +414,13 @@ func NewPlanCacheValue(plan Plan, names []*types.FieldName, srcMap map[*model.Ta // PlanCacheStmt store prepared ast from PrepareExec and other related fields type PlanCacheStmt struct { - PreparedAst *ast.Prepared - StmtDB string // which DB the statement will be processed over - VisitInfos []visitInfo - ColumnInfos interface{} + PreparedAst *ast.Prepared + StmtDB string // which DB the statement will be processed over + VisitInfos []visitInfo + ColumnInfos interface{} + // Executor is only used for point get scene. + // Notice that we should only cache the PointGetExecutor that have a snapshot with MaxTS in it. + // If the current plan is not PointGet or does not use MaxTS optimization, this value should be nil here. Executor interface{} NormalizedSQL string NormalizedPlan string diff --git a/session/session.go b/session/session.go index a8a54ad82b4d9..87e7e70b6995c 100644 --- a/session/session.go +++ b/session/session.go @@ -2197,10 +2197,6 @@ func (s *session) ExecuteStmt(ctx context.Context, stmtNode ast.StmtNode) (sqlex // Transform abstract syntax tree to a physical plan(stored in executor.ExecStmt). compiler := executor.Compiler{Ctx: s} stmt, err := compiler.Compile(ctx, stmtNode) - if err == nil { - err = sessiontxn.OptimizeWithPlanAndThenWarmUp(s, stmt.Plan) - } - if err != nil { s.rollbackOnError(ctx) diff --git a/sessiontxn/interface.go b/sessiontxn/interface.go index 85d9217a90c0f..d91f1c291b588 100644 --- a/sessiontxn/interface.go +++ b/sessiontxn/interface.go @@ -160,8 +160,10 @@ type TxnManager interface { // GetReadReplicaScope returns the read replica scope GetReadReplicaScope() string // GetStmtReadTS returns the read timestamp used by select statement (not for select ... for update) + // Calling this method will activate the txn implicitly if current read is not stale/historical read GetStmtReadTS() (uint64, error) // GetStmtForUpdateTS returns the read timestamp used by update/insert/delete or select ... for update + // Calling this method will activate the txn implicitly if current read is not stale/historical read GetStmtForUpdateTS() (uint64, error) // GetContextProvider returns the current TxnContextProvider GetContextProvider() TxnContextProvider diff --git a/sessiontxn/isolation/optimistic.go b/sessiontxn/isolation/optimistic.go index 3c60eba09331b..9a1f8d58aabbd 100644 --- a/sessiontxn/isolation/optimistic.go +++ b/sessiontxn/isolation/optimistic.go @@ -114,6 +114,12 @@ func (p *OptimisticTxnContextProvider) AdviseOptimizeWithPlan(plan interface{}) return nil } + if p.txn != nil { + // `p.txn != nil` means the txn has already been activated, we should not optimize the startTS because the startTS + // has already been used. + return nil + } + realPlan, ok := plan.(plannercore.Plan) if !ok { return nil @@ -141,7 +147,7 @@ func (p *OptimisticTxnContextProvider) AdviseOptimizeWithPlan(plan interface{}) zap.Uint64("conn", sessVars.ConnectionID), zap.String("text", sessVars.StmtCtx.OriginalSQL), ) - return nil + return err } p.optimizeWithMaxTS = true