From 3a5f434aad408299d58ab34bbb73af2ac0b3a5bf Mon Sep 17 00:00:00 2001 From: cfzjywxk Date: Wed, 14 Jul 2021 10:57:32 +0800 Subject: [PATCH] txn: change lock into put record for unique index key lock (#25730) --- executor/batch_point_get.go | 35 ++++++++++++++++++++++++ executor/point_get.go | 11 ++++++++ session/pessimistic_test.go | 54 +++++++++++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+) diff --git a/executor/batch_point_get.go b/executor/batch_point_get.go index 93778d76bc1ff..0d8c355319d8a 100644 --- a/executor/batch_point_get.go +++ b/executor/batch_point_get.go @@ -409,6 +409,23 @@ func (e *BatchPointGetExec) initialize(ctx context.Context) error { if err != nil { return err } + // Change the unique index LOCK into PUT record. + if len(indexKeys) > 0 { + if !e.txn.Valid() { + return kv.ErrInvalidTxn + } + membuf := e.txn.GetMemBuffer() + for _, idxKey := range indexKeys { + handleVal := handleVals[string(idxKey)] + if len(handleVal) == 0 { + continue + } + err = membuf.Set(idxKey, handleVal) + if err != nil { + return err + } + } + } } // Fetch all values. values, err = batchGetter.BatchGet(ctx, keys) @@ -420,6 +437,7 @@ func (e *BatchPointGetExec) initialize(ctx context.Context) error { if e.lock && rc { existKeys = make([]kv.Key, 0, 2*len(values)) } + changeLockToPutIdxKeys := make([]kv.Key, 0, len(indexKeys)) e.values = make([][]byte, 0, len(values)) for i, key := range keys { val := values[string(key)] @@ -439,6 +457,7 @@ func (e *BatchPointGetExec) initialize(ctx context.Context) error { // lock primary key for clustered index table is redundant if len(indexKeys) != 0 { existKeys = append(existKeys, indexKeys[i]) + changeLockToPutIdxKeys = append(changeLockToPutIdxKeys, indexKeys[i]) } } } @@ -448,6 +467,22 @@ func (e *BatchPointGetExec) initialize(ctx context.Context) error { if err != nil { return err } + if len(changeLockToPutIdxKeys) > 0 { + if !e.txn.Valid() { + return kv.ErrInvalidTxn + } + for _, idxKey := range changeLockToPutIdxKeys { + membuf := e.txn.GetMemBuffer() + handleVal := handleVals[string(idxKey)] + if len(handleVal) == 0 { + return kv.ErrNotExist + } + err = membuf.Set(idxKey, handleVal) + if err != nil { + return err + } + } + } } e.handles = handles return nil diff --git a/executor/point_get.go b/executor/point_get.go index 83678d29330c0..0c07b365bdeb0 100644 --- a/executor/point_get.go +++ b/executor/point_get.go @@ -247,6 +247,17 @@ func (e *PointGetExecutor) Next(ctx context.Context, req *chunk.Chunk) error { if err != nil { return err } + // Change the unique index LOCK into PUT record. + if e.lock && len(e.handleVal) > 0 { + if !e.txn.Valid() { + return kv.ErrInvalidTxn + } + memBuffer := e.txn.GetMemBuffer() + err = memBuffer.Set(e.idxKey, e.handleVal) + if err != nil { + return err + } + } } if len(e.handleVal) == 0 { return nil diff --git a/session/pessimistic_test.go b/session/pessimistic_test.go index 2da876344a93e..fd5a59d7dc786 100644 --- a/session/pessimistic_test.go +++ b/session/pessimistic_test.go @@ -2595,3 +2595,57 @@ func (s *testPessimisticSuite) TestAsyncCommitCalTSFail(c *C) { tk2.MustExec("update tk set c2 = c2 + 1") tk2.MustExec("commit") } + +func (s *testPessimisticSuite) TestChangeLockToPut(c *C) { + tk := testkit.NewTestKitWithInit(c, s.store) + tk2 := testkit.NewTestKitWithInit(c, s.store) + + tk.MustExec("use test") + tk2.MustExec("use test") + tk.MustExec("drop table if exists tk") + tk.MustExec("create table t1(c1 varchar(20) key, c2 int, c3 int, unique key k1(c2), key k2(c3))") + tk.MustExec(`insert into t1 values ("1", 1, 1), ("2", 2, 2), ("3", 3, 3)`) + + // Test point get change lock to put. + for _, mode := range []string{"REPEATABLE-READ", "READ-COMMITTED"} { + tk.MustExec(fmt.Sprintf(`set tx_isolation = "%s"`, mode)) + tk.MustExec("begin pessimistic") + tk.MustQuery(`select * from t1 where c1 = "1" for update`).Check(testkit.Rows("1 1 1")) + tk.MustExec("commit") + tk.MustExec("begin pessimistic") + tk.MustQuery(`select * from t1 where c1 = "1" for update`).Check(testkit.Rows("1 1 1")) + tk.MustExec("commit") + tk.MustExec("admin check table t1") + tk2.MustExec("begin") + tk2.MustQuery(`select * from t1 use index(k1) where c2 = "1" for update`).Check(testkit.Rows("1 1 1")) + tk2.MustQuery(`select * from t1 use index(k1) where c2 = "3" for update`).Check(testkit.Rows("3 3 3")) + tk2.MustExec("commit") + tk2.MustExec("begin") + tk2.MustQuery(`select * from t1 use index(k2) where c3 = 1`).Check(testkit.Rows("1 1 1")) + tk2.MustQuery("select * from t1 use index(k2) where c3 > 1").Check(testkit.Rows("2 2 2", "3 3 3")) + tk2.MustExec("commit") + } + + // Test batch point get change lock to put. + for _, mode := range []string{"REPEATABLE-READ", "READ-COMMITTED"} { + tk.MustExec(fmt.Sprintf(`set tx_isolation = "%s"`, mode)) + tk.MustExec("begin pessimistic") + tk.MustQuery(`select * from t1 where c1 in ("1", "5", "3") for update`).Check(testkit.Rows("1 1 1", "3 3 3")) + tk.MustExec("commit") + tk.MustExec("begin pessimistic") + tk.MustQuery(`select * from t1 where c1 in ("1", "2", "8") for update`).Check(testkit.Rows("1 1 1", "2 2 2")) + tk.MustExec("commit") + tk.MustExec("admin check table t1") + tk2.MustExec("begin") + tk2.MustQuery(`select * from t1 use index(k1) where c2 in ("1", "2", "3") for update`).Check(testkit.Rows("1 1 1", "2 2 2", "3 3 3")) + tk2.MustQuery(`select * from t1 use index(k2) where c2 in ("2") for update`).Check(testkit.Rows("2 2 2")) + tk2.MustExec("commit") + tk2.MustExec("begin") + tk2.MustQuery(`select * from t1 use index(k2) where c3 in (5, 8)`).Check(testkit.Rows()) + tk2.MustQuery(`select * from t1 use index(k2) where c3 in (1, 8) for update`).Check(testkit.Rows("1 1 1")) + tk2.MustQuery(`select * from t1 use index(k2) where c3 > 1`).Check(testkit.Rows("2 2 2", "3 3 3")) + tk2.MustExec("commit") + } + + tk.MustExec("admin check table t1") +}