cop: ignore locks with the same start_ts (#52461)

close #52460

pkg/store/copr/coprocessor.go

@@ -190,6 +190,17 @@ func (c *CopClient) BuildCopIterator(ctx context.Context, req *kv.Request, vars
 		buildTaskElapsed: *buildOpt.elapsed,
 		runawayChecker:   req.RunawayChecker,
 	}
+	// Pipelined-dml can flush locks when it is still reading.
+	// The coprocessor of the txn should not be blocked by itself.
+	// It should be the only case where a coprocessor can read locks of the same ts.
+	//
+	// But when start_ts is not obtained from PD,
+	// the start_ts could conflict with another pipelined-txn's start_ts.
+	// in which case the locks of same ts cannot be ignored.
+	// We rely on the assumption: start_ts is not from PD => this is a stale read.
+	if !req.IsStaleness {
+		it.resolvedLocks.Put(req.StartTs)
+	}
 	it.tasks = tasks
 	if it.concurrency > len(tasks) {
 		it.concurrency = len(tasks)

pkg/store/mockstore/unistore/cophandler/closure_exec.go

@@ -1185,7 +1185,7 @@ func safeCopy(b []byte) []byte {
 }
 
 func checkLock(lock mvcc.Lock, key []byte, startTS uint64, resolved []uint64) error {
-	if isResolved(startTS, resolved) {
+	if isResolved(lock.StartTS, resolved) {
 		return nil
 	}
 	lockVisible := lock.StartTS < startTS