*: fix condition check push down for pessimistic transaction (#14141) (…

…#14175)

session/pessimistic_test.go

@@ -595,6 +595,22 @@ func (s *testPessimisticSuite) TestInnodbLockWaitTimeout(c *C) {
 	tk.MustExec("drop table if exists tk")
 }
 
+func (s *testPessimisticSuite) TestPushConditionCheckForPessimisticTxn(c *C) {
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	tk1 := testkit.NewTestKitWithInit(c, s.store)
+	defer tk.MustExec("drop table if exists t")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (i int key)")
+	tk.MustExec("insert into t values (1)")
+
+	tk.MustExec("set tidb_txn_mode = 'pessimistic'")
+	tk.MustExec("begin")
+	tk1.MustExec("delete from t where i = 1")
+	tk.MustExec("insert into t values (1) on duplicate key update i = values(i)")
+	tk.MustExec("commit")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1"))
+}
+
 func (s *testPessimisticSuite) TestInnodbLockWaitTimeoutWaitStart(c *C) {
 	// prepare work
 	tk := testkit.NewTestKitWithInit(c, s.store)

store/mockstore/mocktikv/mvcc_leveldb.go

@@ -628,6 +628,7 @@ func (mvcc *MVCCLevelDB) Prewrite(req *kvrpcpb.PrewriteRequest) []error {
 	mutations := req.Mutations
 	primary := req.PrimaryLock
 	startTS := req.StartVersion
+	forUpdateTS := req.GetForUpdateTs()
 	ttl := req.LockTtl
 	mvcc.mu.Lock()
 	defer mvcc.mu.Unlock()
@@ -639,7 +640,8 @@ func (mvcc *MVCCLevelDB) Prewrite(req *kvrpcpb.PrewriteRequest) []error {
 	for i, m := range mutations {
 		// If the operation is Insert, check if key is exists at first.
 		var err error
-		if m.GetOp() == kvrpcpb.Op_Insert {
+		// no need to check insert values for pessimistic transaction.
+		if m.GetOp() == kvrpcpb.Op_Insert && forUpdateTS == 0 {
 			v, err := mvcc.getValue(m.Key, startTS, kvrpcpb.IsolationLevel_SI)
 			if err != nil {
 				errs = append(errs, err)