cherry pick pingcap#35732 to release-6.1

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

executor/executor.go

@@ -1025,6 +1025,15 @@ func (e *SelectLockExec) Next(ctx context.Context, req *chunk.Chunk) error {
 					physTblID := tblID
 					if physTblColIdx, ok := e.tblID2PhysTblIDColIdx[tblID]; ok {
 						physTblID = row.GetInt64(physTblColIdx)
+						if physTblID == 0 {
+							// select * from t1 left join t2 on t1.c = t2.c for update
+							// The join right side might be added NULL in left join
+							// In that case, physTblID is 0, so skip adding the lock.
+							//
+							// Note, we can't distinguish whether it's the left join case,
+							// or a bug that TiKV return without correct physical ID column.
+							continue
+						}
 					}
 					e.keys = append(e.keys, tablecodec.EncodeRowKeyWithHandle(physTblID, handle))
 				}

executor/union_scan_test.go

@@ -15,11 +15,13 @@
 package executor_test
 
 import (
+	"encoding/hex"
 	"fmt"
 	"testing"
 	"time"
 
 	"github.com/pingcap/tidb/executor"
+	"github.com/pingcap/tidb/store/helper"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/util/benchdaily"
@@ -456,6 +458,28 @@ func TestIssue28073(t *testing.T) {
 		break
 	}
 	require.False(t, exist)
+
+	// Another case, left join on partition table should not generate locks on physical ID = 0
+	tk.MustExec("drop table if exists t1, t2;")
+	tk.MustExec("create table t1  (c_int int, c_str varchar(40), primary key (c_int, c_str));")
+	tk.MustExec("create table t2  (c_int int, c_str varchar(40), primary key (c_int)) partition by hash (c_int) partitions 4;")
+	tk.MustExec("insert into t1 (`c_int`, `c_str`) values (1, 'upbeat solomon'), (5, 'sharp rubin');")
+	tk.MustExec("insert into t2 (`c_int`, `c_str`) values (1, 'clever haibt'), (4, 'kind margulis');")
+	tk.MustExec("begin pessimistic;")
+	tk.MustQuery("select * from t1 left join t2 on t1.c_int = t2.c_int for update;").Check(testkit.Rows(
+		"1 upbeat solomon 1 clever haibt",
+		"5 sharp rubin <nil> <nil>",
+	))
+	key, err := hex.DecodeString("7480000000000000005F728000000000000000")
+	require.NoError(t, err)
+	h := helper.NewHelper(store.(helper.Storage))
+	resp, err := h.GetMvccByEncodedKey(key)
+	require.NoError(t, err)
+	require.Nil(t, resp.Info.Lock)
+	require.Len(t, resp.Info.Writes, 0)
+	require.Len(t, resp.Info.Values, 0)
+
+	tk.MustExec("rollback;")
 }
 
 func TestIssue32422(t *testing.T) {