*: avoid pass session ID by context & fix no schemaChecker error log

executor/adapter.go

@@ -306,7 +306,6 @@ func (a *ExecStmt) Exec(ctx context.Context) (_ sqlexec.RecordSet, err error) {
 	}()
 
 	sctx := a.Ctx
-	ctx = util.SetSessionID(ctx, sctx.GetSessionVars().ConnectionID)
 	if _, ok := a.Plan.(*plannercore.Analyze); ok && sctx.GetSessionVars().InRestrictedSQL {
 		oriStats, _ := sctx.GetSessionVars().GetSystemVar(variable.TiDBBuildStatsConcurrency)
 		oriScan := sctx.GetSessionVars().DistSQLScanConcurrency()
@@ -582,6 +581,7 @@ func (a *ExecStmt) handlePessimisticDML(ctx context.Context, e Executor) error {
 		var lockKeyStats *util.LockKeysDetails
 		ctx = context.WithValue(ctx, util.LockKeysDetailCtxKey, &lockKeyStats)
 		startLocking := time.Now()
+		txn.SetOption(kv.SessionID, seVars.ConnectionID)
 		err = txn.LockKeys(ctx, lockCtx, keys...)
 		if lockKeyStats != nil {
 			seVars.StmtCtx.MergeLockKeysExecDetails(lockKeyStats)

executor/executor.go

@@ -1002,7 +1002,8 @@ func doLockKeys(ctx context.Context, se sessionctx.Context, lockCtx *tikvstore.L
 	}
 	var lockKeyStats *tikvutil.LockKeysDetails
 	ctx = context.WithValue(ctx, tikvutil.LockKeysDetailCtxKey, &lockKeyStats)
-	err = txn.LockKeys(tikvutil.SetSessionID(ctx, se.GetSessionVars().ConnectionID), lockCtx, keys...)
+	txn.SetOption(kv.SessionID, se.GetSessionVars().ConnectionID)
+	err = txn.LockKeys(ctx, lockCtx, keys...)
 	if lockKeyStats != nil {
 		sctx.MergeLockKeysExecDetails(lockKeyStats)
 	}

executor/simple.go

@@ -40,7 +40,6 @@ import (
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/tikv/oracle"
-	tikvutil "github.com/pingcap/tidb/store/tikv/util"
 	"github.com/pingcap/tidb/types"
 	driver "github.com/pingcap/tidb/types/parser_driver"
 	"github.com/pingcap/tidb/util"
@@ -949,7 +948,8 @@ func (e *SimpleExec) executeAlterUser(s *ast.AlterUserStmt) error {
 		if err != nil {
 			return err
 		}
-		err = txn.Commit(tikvutil.SetSessionID(context.TODO(), e.ctx.GetSessionVars().ConnectionID))
+		txn.SetOption(kv.SessionID, e.ctx.GetSessionVars().ConnectionID)
+		err = txn.Commit(context.TODO())
 		if err != nil {
 			return err
 		}

go.sum

@@ -500,6 +500,7 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/sasha-s/go-deadlock v0.2.0/go.mod h1:StQn567HiB1fF2yJ44N9au7wOhrPS3iZqiDbRupzT10=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shirou/gopsutil v2.19.10+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/shirou/gopsutil v3.21.2+incompatible h1:U+YvJfjCh6MslYlIAXvPtzhW3YZEtc9uncueUNpD/0A=

kv/option.go

@@ -59,6 +59,8 @@ const (
 	IsStalenessReadOnly
 	// MatchStoreLabels indicates the labels the store should be matched
 	MatchStoreLabels
+	// SessionID store the session (connection) ID to the transaction.
+	SessionID
 )
 
 // ReplicaReadType is the type of replica to read data from

session/session.go

@@ -518,6 +518,7 @@ func (s *session) doCommit(ctx context.Context) error {
 	}
 	s.txn.SetOption(kv.EnableAsyncCommit, s.GetSessionVars().EnableAsyncCommit)
 	s.txn.SetOption(kv.Enable1PC, s.GetSessionVars().Enable1PC)
+	s.txn.SetOption(kv.SessionID, s.GetSessionVars().ConnectionID)
 	// priority of the sysvar is lower than `start transaction with causal consistency only`
 	if val := s.txn.GetOption(kv.GuaranteeLinearizability); val == nil || val.(bool) {
 		// We needn't ask the TiKV client to guarantee linearizability for auto-commit transactions
@@ -529,7 +530,7 @@ func (s *session) doCommit(ctx context.Context) error {
 			s.GetSessionVars().TxnCtx.IsExplicit && s.GetSessionVars().GuaranteeLinearizability)
 	}
 
-	return s.txn.Commit(tikvutil.SetSessionID(ctx, s.GetSessionVars().ConnectionID))
+	return s.txn.Commit(ctx)
 }
 
 // removeTempTableFromBuffer filters out the temporary table key-values.

store/driver/txn/txn_driver.go

@@ -170,6 +170,8 @@ func (txn *tikvTxn) SetOption(opt int, val interface{}) {
 		txn.KVTxn.GetSnapshot().SetIsStatenessReadOnly(val.(bool))
 	case kv.MatchStoreLabels:
 		txn.KVTxn.GetSnapshot().SetMatchStoreLabels(val.([]*metapb.StoreLabel))
+	case kv.SessionID:
+		txn.KVTxn.SetSessionID(val.(uint64))
 	}
 }
 

store/tikv/tests/1pc_test.go

@@ -20,13 +20,13 @@ import (
 	"github.com/pingcap/tidb/store/tikv"
 	"github.com/pingcap/tidb/store/tikv/metrics"
 	"github.com/pingcap/tidb/store/tikv/oracle"
-	"github.com/pingcap/tidb/store/tikv/util"
 )
 
 func (s *testAsyncCommitCommon) begin1PC(c *C) tikv.TxnProbe {
 	txn, err := s.store.Begin()
 	c.Assert(err, IsNil)
 	txn.SetEnable1PC(true)
+	txn.SetSessionID(1)
 	return tikv.TxnProbe{KVTxn: txn}
 }
 
@@ -44,7 +44,7 @@ func (s *testOnePCSuite) SetUpTest(c *C) {
 }
 
 func (s *testOnePCSuite) Test1PC(c *C) {
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 
 	k1 := []byte("k1")
 	v1 := []byte("v1")
@@ -67,20 +67,21 @@ func (s *testOnePCSuite) Test1PC(c *C) {
 	txn = s.begin1PC(c)
 	err = txn.Set(k2, v2)
 	c.Assert(err, IsNil)
+	txn.SetSessionID(0)
 	err = txn.Commit(context.Background())
 	c.Assert(err, IsNil)
 	c.Assert(txn.GetCommitter().IsOnePC(), IsFalse)
 	c.Assert(txn.GetCommitter().GetOnePCCommitTS(), Equals, uint64(0))
 	c.Assert(txn.GetCommitter().GetCommitTS(), Greater, txn.StartTS())
 
 	// 1PC doesn't work if system variable not set
-
 	k3 := []byte("k3")
 	v3 := []byte("v3")
 
 	txn = s.begin(c)
 	err = txn.Set(k3, v3)
 	c.Assert(err, IsNil)
+	txn.SetSessionID(1)
 	err = txn.Commit(ctx)
 	c.Assert(err, IsNil)
 	c.Assert(txn.GetCommitter().IsOnePC(), IsFalse)
@@ -142,7 +143,7 @@ func (s *testOnePCSuite) Test1PC(c *C) {
 }
 
 func (s *testOnePCSuite) Test1PCIsolation(c *C) {
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 
 	k := []byte("k")
 	v1 := []byte("v1")
@@ -183,7 +184,7 @@ func (s *testOnePCSuite) Test1PCDisallowMultiRegion(c *C) {
 		return
 	}
 
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 
 	txn := s.begin1PC(c)
 
@@ -234,7 +235,7 @@ func (s *testOnePCSuite) Test1PCLinearizability(c *C) {
 	c.Assert(err, IsNil)
 	err = t2.Set([]byte("b"), []byte("b1"))
 	c.Assert(err, IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	// t2 commits earlier than t1
 	err = t2.Commit(ctx)
 	c.Assert(err, IsNil)
@@ -255,7 +256,7 @@ func (s *testOnePCSuite) Test1PCWithMultiDC(c *C) {
 	err := localTxn.Set([]byte("a"), []byte("a1"))
 	localTxn.SetScope("bj")
 	c.Assert(err, IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	err = localTxn.Commit(ctx)
 	c.Assert(err, IsNil)
 	c.Assert(localTxn.GetCommitter().IsOnePC(), IsFalse)
@@ -276,7 +277,7 @@ func (s *testOnePCSuite) TestTxnCommitCounter(c *C) {
 	txn := s.begin(c)
 	err := txn.Set([]byte("k"), []byte("v"))
 	c.Assert(err, IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	err = txn.Commit(ctx)
 	c.Assert(err, IsNil)
 	curr := metrics.GetTxnCommitCounter()

store/tikv/tests/async_commit_fail_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/store/tikv"
 	tikverr "github.com/pingcap/tidb/store/tikv/error"
-	"github.com/pingcap/tidb/store/tikv/util"
 )
 
 type testAsyncCommitFailSuite struct {
@@ -57,7 +56,7 @@ func (s *testAsyncCommitFailSuite) TestFailAsyncCommitPrewriteRpcErrors(c *C) {
 	t1 := s.beginAsyncCommit(c)
 	err := t1.Set([]byte("a"), []byte("a1"))
 	c.Assert(err, IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	err = t1.Commit(ctx)
 	c.Assert(err, NotNil)
 	c.Assert(terror.ErrorEqual(err, terror.ErrResultUndetermined), IsTrue, Commentf("%s", errors.ErrorStack(err)))
@@ -99,7 +98,7 @@ func (s *testAsyncCommitFailSuite) TestAsyncCommitPrewriteCancelled(c *C) {
 	c.Assert(err, IsNil)
 	err = t1.Set([]byte("z"), []byte("z"))
 	c.Assert(err, IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	err = t1.Commit(ctx)
 	c.Assert(err, NotNil)
 	_, ok := errors.Cause(err).(*tikverr.ErrWriteConflict)
@@ -117,7 +116,7 @@ func (s *testAsyncCommitFailSuite) TestPointGetWithAsyncCommit(c *C) {
 
 	// PointGet cannot ignore async commit transactions' locks.
 	c.Assert(failpoint.Enable("github.com/pingcap/tidb/store/tikv/asyncCommitDoNothing", "return"), IsNil)
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
 	err := txn.Commit(ctx)
 	c.Assert(err, IsNil)
 	c.Assert(txn.GetCommitter().IsAsyncCommit(), IsTrue)
@@ -167,9 +166,10 @@ func (s *testAsyncCommitFailSuite) TestSecondaryListInPrimaryLock(c *C) {
 	var sessionID uint64 = 0
 	test := func(keys []string, values []string) {
 		sessionID++
-		ctx := context.WithValue(context.Background(), util.SessionID, sessionID)
+		ctx := context.Background()
 
 		txn := s.beginAsyncCommit(c)
+		txn.SetSessionID(sessionID)
 		for i := range keys {
 			txn.Set([]byte(keys[i]), []byte(values[i]))
 		}
@@ -228,7 +228,8 @@ func (s *testAsyncCommitFailSuite) TestAsyncCommitContextCancelCausingUndetermin
 		c.Assert(failpoint.Disable("github.com/pingcap/tidb/store/tikv/rpcContextCancelErr"), IsNil)
 	}()
 
-	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
+	ctx := context.Background()
+	txn.SetSessionID(1)
 	err = txn.Commit(ctx)
 	c.Assert(err, NotNil)
 	c.Assert(txn.GetCommitter().GetUndeterminedErr(), NotNil)

store/tikv/tests/async_commit_test.go

@@ -134,6 +134,7 @@ func (s *testAsyncCommitCommon) beginAsyncCommit(c *C) tikv.TxnProbe {
 	txn, err := s.store.Begin()
 	c.Assert(err, IsNil)
 	txn.SetEnableAsyncCommit(true)
+	txn.SetSessionID(1)
 	return tikv.TxnProbe{KVTxn: txn}
 }
 

store/tikv/txn.go

@@ -88,6 +88,7 @@ type KVTxn struct {
 	causalConsistency  bool
 	scope              string
 	kvFilter           KVFilter
+	sessionID          uint64
 }
 
 func extractStartTs(store *KVStore, options kv.TransactionOption) (uint64, error) {
@@ -285,6 +286,11 @@ func (txn *KVTxn) SetKVFilter(filter KVFilter) {
 	txn.kvFilter = filter
 }
 
+// SetSessionID sets the session ID for this transaction.
+func (txn *KVTxn) SetSessionID(id uint64) {
+	txn.sessionID = id
+}
+
 // IsPessimistic returns true if it is pessimistic.
 func (txn *KVTxn) IsPessimistic() bool {
 	return txn.isPessimistic
@@ -325,18 +331,11 @@ func (txn *KVTxn) Commit(ctx context.Context) error {
 	start := time.Now()
 	defer func() { metrics.TxnCmdHistogramWithCommit.Observe(time.Since(start).Seconds()) }()
 
-	// sessionID is used for log.
-	var sessionID uint64
-	val := ctx.Value(util.SessionID)
-	if val != nil {
-		sessionID = val.(uint64)
-	}
-
 	var err error
 	// If the txn use pessimistic lock, committer is initialized.
 	committer := txn.committer
 	if committer == nil {
-		committer, err = newTwoPhaseCommitter(txn, sessionID)
+		committer, err = newTwoPhaseCommitter(txn, txn.sessionID)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -369,9 +368,7 @@ func (txn *KVTxn) Commit(ctx context.Context) error {
 	// pessimistic transaction should also bypass latch.
 	if txn.store.txnLatches == nil || txn.IsPessimistic() {
 		err = committer.execute(ctx)
-		if val == nil || sessionID > 0 {
-			txn.onCommitted(err)
-		}
+		txn.onCommitted(err)
 		logutil.Logger(ctx).Debug("[kv] txnLatches disabled, 2pc directly", zap.Error(err))
 		return errors.Trace(err)
 	}
@@ -390,9 +387,7 @@ func (txn *KVTxn) Commit(ctx context.Context) error {
 		return &tikverr.ErrWriteConflictInLatch{StartTS: txn.startTS}
 	}
 	err = committer.execute(ctx)
-	if val == nil || sessionID > 0 {
-		txn.onCommitted(err)
-	}
+	txn.onCommitted(err)
 	if err == nil {
 		lock.SetCommitTS(committer.commitTS)
 	}
@@ -548,14 +543,7 @@ func (txn *KVTxn) LockKeys(ctx context.Context, lockCtx *tikv.LockCtx, keysInput
 	keys = deduplicateKeys(keys)
 	if txn.IsPessimistic() && lockCtx.ForUpdateTS > 0 {
 		if txn.committer == nil {
-			// sessionID is used for log.
-			var sessionID uint64
-			var err error
-			val := ctx.Value(util.SessionID)
-			if val != nil {
-				sessionID = val.(uint64)
-			}
-			txn.committer, err = newTwoPhaseCommitter(txn, sessionID)
+			txn.committer, err = newTwoPhaseCommitter(txn, txn.sessionID)
 			if err != nil {
 				return err
 			}