Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

executor: revert #25730 to fix #28011 (#42488) #42502

Closed
wants to merge 1 commit into from
Closed

executor: revert #25730 to fix #28011 (#42488) #42502

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 0 additions & 35 deletions executor/batch_point_get.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -445,23 +445,6 @@ 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)
Expand All @@ -473,7 +456,6 @@ 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)]
Expand Down Expand Up @@ -508,7 +490,6 @@ 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])
}
}
}
Expand All @@ -518,22 +499,6 @@ 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
Expand Down
128 changes: 85 additions & 43 deletions tests/realtikvtest/pessimistictest/pessimistic_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2856,61 +2856,103 @@ func TestAsyncCommitCalTSFail(t *testing.T) {
tk2.MustExec("commit")
}

<<<<<<< HEAD
func TestChangeLockToPut(t *testing.T) {
store, clean := realtikvtest.CreateMockStoreAndSetup(t)
defer clean()
=======
func TestAsyncCommitAndForeignKey(t *testing.T) {
defer config.RestoreFunc()()
config.UpdateGlobal(func(conf *config.Config) {
conf.TiKVClient.AsyncCommit.SafeWindow = time.Second
conf.TiKVClient.AsyncCommit.AllowedClockDrift = 0
})
store := realtikvtest.CreateMockStoreAndSetup(t)
tk := createAsyncCommitTestKit(t, store)
tk.MustExec("drop table if exists t_parent, t_child")
tk.MustExec("create table t_parent (id int primary key)")
tk.MustExec("create table t_child (id int primary key, pid int, foreign key (pid) references t_parent(id) on delete cascade on update cascade)")
tk.MustExec("insert into t_parent values (1),(2),(3),(4)")
tk.MustExec("insert into t_child values (1,1),(2,2),(3,3)")
tk.MustExec("set tidb_enable_1pc = true")
tk.MustExec("begin pessimistic")
tk.MustExec("delete from t_parent where id in (1,4)")
tk.MustExec("update t_parent set id=22 where id=2")
tk.MustExec("commit")
tk.MustQuery("select * from t_parent order by id").Check(testkit.Rows("3", "22"))
tk.MustQuery("select * from t_child order by id").Check(testkit.Rows("2 22", "3 3"))
}

func TestTransactionIsolationAndForeignKey(t *testing.T) {
if !*realtikvtest.WithRealTiKV {
t.Skip("The test only support test with tikv.")
}
store := realtikvtest.CreateMockStoreAndSetup(t)
tk := testkit.NewTestKit(t, store)
tk2 := testkit.NewTestKit(t, store)
tk.MustExec("use test")
tk2.MustExec("use test")
tk.MustExec("drop table if exists t1,t2")
tk.MustExec("create table t1 (id int primary key)")
tk.MustExec("create table t2 (id int primary key, pid int, foreign key (pid) references t1(id) on delete cascade on update cascade)")
tk.MustExec("insert into t1 values (1)")
tk.MustExec("set tx_isolation = 'READ-COMMITTED'")
tk.MustExec("begin pessimistic")
tk.MustExec("insert into t2 values (1,1)")
tk.MustGetDBError("insert into t2 values (2,2)", plannercore.ErrNoReferencedRow2)
tk2.MustExec("insert into t1 values (2)")
tk.MustQuery("select * from t1").Check(testkit.Rows("1", "2"))
tk.MustExec("insert into t2 values (2,2)")
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
tk2.MustExec("delete from t1 where id=2")
}()
time.Sleep(time.Millisecond * 10)
tk.MustExec("commit")
wg.Wait()
tk.MustQuery("select * from t1").Check(testkit.Rows("1"))
tk.MustQuery("select * from t2").Check(testkit.Rows("1 1"))
tk2.MustExec("delete from t1 where id=1")
tk.MustQuery("select * from t1").Check(testkit.Rows())
tk.MustQuery("select * from t2").Check(testkit.Rows())
tk.MustExec("admin check table t1")
tk.MustExec("admin check table t2")
}

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)`)
func TestIssue28011(t *testing.T) {
store := realtikvtest.CreateMockStoreAndSetup(t)
>>>>>>> 273763b51e8 (executor: revert #25730 to fix #28011 (#42488))

// 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")
}
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")

// 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")
for _, tt := range []struct {
name string
lockQuery string
finalRows [][]interface{}
}{
{"Update", "update t set b = 'x' where a = 'a'", testkit.Rows("a x", "b y", "c z")},
{"BatchUpdate", "update t set b = 'x' where a in ('a', 'b', 'c')", testkit.Rows("a x", "b y", "c x")},
{"SelectForUpdate", "select a from t where a = 'a' for update", testkit.Rows("a x", "b y", "c z")},
{"BatchSelectForUpdate", "select a from t where a in ('a', 'b', 'c') for update", testkit.Rows("a x", "b y", "c z")},
} {
t.Run(tt.name, func(t *testing.T) {
tk.MustExec("drop table if exists t")
tk.MustExec("create table t (a varchar(10) primary key nonclustered, b varchar(10))")
tk.MustExec("insert into t values ('a', 'x'), ('b', 'x'), ('c', 'z')")
tk.MustExec("begin")
tk.MustExec(tt.lockQuery)
tk.MustQuery("select a from t").Check(testkit.Rows("a", "b", "c"))
tk.MustExec("replace into t values ('b', 'y')")
tk.MustQuery("select a from t").Check(testkit.Rows("a", "b", "c"))
tk.MustQuery("select a, b from t order by a").Check(tt.finalRows)
tk.MustExec("commit")
tk.MustQuery("select a, b from t order by a").Check(tt.finalRows)
tk.MustExec("admin check table t")
})
}

tk.MustExec("admin check table t1")
}

func createTable(part bool, columnNames []string, columnTypes []string) string {
Expand Down