executor: record lockkeys waited flag and duration in stmtctx (pingcap#14209)

Author: cfzjywxk

executor/adapter.go

@@ -542,12 +542,7 @@ func (a *ExecStmt) handlePessimisticDML(ctx context.Context, e Executor) error {
 			return nil
 		}
 		seVars := sctx.GetSessionVars()
-		lockCtx := &kv.LockCtx{
-			Killed:        &seVars.Killed,
-			ForUpdateTS:   txnCtx.GetForUpdateTS(),
-			LockWaitTime:  seVars.LockWaitTimeout,
-			WaitStartTime: seVars.StmtCtx.GetLockWaitStartTime(),
-		}
+		lockCtx := newLockCtx(seVars, seVars.LockWaitTimeout)
 		err = txn.LockKeys(ctx, lockCtx, keys...)
 		if err == nil {
 			return nil

executor/executor.go

@@ -826,10 +826,12 @@ func (e *SelectLockExec) Next(ctx context.Context, req *chunk.Chunk) error {
 
 func newLockCtx(seVars *variable.SessionVars, lockWaitTime int64) *kv.LockCtx {
 	return &kv.LockCtx{
-		Killed:        &seVars.Killed,
-		ForUpdateTS:   seVars.TxnCtx.GetForUpdateTS(),
-		LockWaitTime:  lockWaitTime,
-		WaitStartTime: seVars.StmtCtx.GetLockWaitStartTime(),
+		Killed:                &seVars.Killed,
+		ForUpdateTS:           seVars.TxnCtx.GetForUpdateTS(),
+		LockWaitTime:          lockWaitTime,
+		WaitStartTime:         seVars.StmtCtx.GetLockWaitStartTime(),
+		PessimisticLockWaited: &seVars.StmtCtx.PessimisticLockWaited,
+		LockKeysDuration:      &seVars.StmtCtx.LockKeysDuration,
 	}
 }
 

kv/kv.go

@@ -198,8 +198,8 @@ type LockCtx struct {
 	ForUpdateTS           uint64
 	LockWaitTime          int64
 	WaitStartTime         time.Time
-	PessimisticLockWaited int32
-	LockTimeWaited        time.Duration
+	PessimisticLockWaited *int32
+	LockKeysDuration      *time.Duration
 }
 
 // Client is used to send request to KV layer.

sessionctx/stmtctx/stmtctx.go

@@ -135,11 +135,13 @@ type StatementContext struct {
 		digest     string
 	}
 	// planNormalized use for cache the normalized plan, avoid duplicate builds.
-	planNormalized    string
-	planDigest        string
-	Tables            []TableEntry
-	PointExec         bool       // for point update cached execution, Constant expression need to set "paramMarker"
-	lockWaitStartTime *time.Time // LockWaitStartTime stores the pessimistic lock wait start time
+	planNormalized        string
+	planDigest            string
+	Tables                []TableEntry
+	PointExec             bool       // for point update cached execution, Constant expression need to set "paramMarker"
+	lockWaitStartTime     *time.Time // LockWaitStartTime stores the pessimistic lock wait start time
+	PessimisticLockWaited int32
+	LockKeysDuration      time.Duration
 }
 
 // StmtHints are SessionVars related sql hints.

store/tikv/2pc.go

@@ -767,7 +767,9 @@ func (action actionPessimisticLock) handleSingleBatch(c *twoPhaseCommitter, bo *
 					return errors.Trace(ErrLockWaitTimeout)
 				}
 			}
-			atomic.StoreInt32(&action.LockCtx.PessimisticLockWaited, 1)
+			if action.LockCtx.PessimisticLockWaited != nil {
+				atomic.StoreInt32(action.LockCtx.PessimisticLockWaited, 1)
+			}
 		}
 
 		// Handle the killed flag when waiting for the pessimistic lock.

store/tikv/txn.go

@@ -372,9 +372,12 @@ func (txn *tikvTxn) LockKeys(ctx context.Context, lockCtx *kv.LockCtx, keysInput
 	var err error
 	defer func() {
 		if err == nil {
-			if lockCtx.PessimisticLockWaited > 0 {
-				lockCtx.LockTimeWaited = time.Since(lockCtx.WaitStartTime)
-				metrics.TiKVPessimisticLockKeysDuration.Observe(lockCtx.LockTimeWaited.Seconds())
+			if lockCtx.PessimisticLockWaited != nil {
+				if atomic.LoadInt32(lockCtx.PessimisticLockWaited) > 0 {
+					timeWaited := time.Since(lockCtx.WaitStartTime)
+					*lockCtx.LockKeysDuration = timeWaited
+					metrics.TiKVPessimisticLockKeysDuration.Observe(timeWaited.Seconds())
+				}
 			}
 		}
 	}()