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

executor: lock duplicated keys on insert-ignore & replace-nothing #42210

Merged
merged 4 commits into from
Mar 15, 2023
Merged

executor: lock duplicated keys on insert-ignore & replace-nothing #42210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cb49291
executor: lock duplicated keys on insert-ingore & replace-nothing
zyguan Mar 14, 2023
74dda1a
fix unit test
zyguan Mar 14, 2023
d487490
Update executor/insert_common.go
zyguan Mar 14, 2023
b4cda16
Update executor/insert_common.go
zyguan Mar 14, 2023
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
12 changes: 12 additions & 0 deletions executor/insert_common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1207,6 +1207,10 @@ func (e *InsertValues) batchCheckAndInsert(ctx context.Context, rows [][]types.D
}
} else {
e.ctx.GetSessionVars().StmtCtx.AppendWarning(r.handleKey.dupErr)
if txnCtx := e.ctx.GetSessionVars().TxnCtx; txnCtx.IsPessimistic {
// lock duplicated row key on insert-ignore
txnCtx.AddUnchangedRowKey(r.handleKey.newKey)
}
continue
}
} else if !kv.IsErrNotFound(err) {
Expand Down Expand Up @@ -1239,6 +1243,10 @@ func (e *InsertValues) batchCheckAndInsert(ctx context.Context, rows [][]types.D
} else {
// If duplicate keys were found in BatchGet, mark row = nil.
e.ctx.GetSessionVars().StmtCtx.AppendWarning(uk.dupErr)
if txnCtx := e.ctx.GetSessionVars().TxnCtx; txnCtx.IsPessimistic {
// lock duplicated unique key on insert-ignore
txnCtx.AddUnchangedRowKey(uk.newKey)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems mysql only locks unique keys as well.

/* init */ select version();
-- init >> +-----------+
-- init    | version() |
-- init    +-----------+
-- init    | 8.0.28    |
-- init    +-----------+
/* init */ drop table if exists t;
-- init >> 0 rows affected
/* init */ create table t (a int primary key, b int unique key, c int);
-- init >> 0 rows affected
/* init */ insert into t values (1, 1, 1);
-- init >> 1 rows affected
/* t1 */ begin;
-- t1 >> 0 rows affected
/* t2 */ begin;
-- t2 >> 0 rows affected
/* t1 */ insert ignore into t values (2, 1, 1);
-- t1 >> 0 rows affected
/* t2 */ select b from t where a = 1 for update;
-- t2 >> +---+
-- t2    | b |
-- t2    +---+
-- t2    | 1 |
-- t2    +---+
/* t2 */ select a from t where b = 1 for update;
-- t2 >> blocked
/* t1 */ commit;
-- t1 >> 0 rows affected
-- t2 >> resumed
-- t2 >> +---+
-- t2    | a |
-- t2    +---+
-- t2    | 1 |
-- t2    +---+
/* t2 */ commit;
-- t2 >> 0 rows affected

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The behaviors of this part are not described in the documents yet, we could add them there.

}
skip = true
break
}
Expand Down Expand Up @@ -1294,6 +1302,10 @@ func (e *InsertValues) removeRow(
if inReplace {
e.ctx.GetSessionVars().StmtCtx.AddAffectedRows(1)
}
_, err := appendUnchangedRowForLock(e.ctx, r.t, handle, oldRow)
if err != nil {
return false, err
}
return true, nil
}

Expand Down
66 changes: 66 additions & 0 deletions executor/insert_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1509,3 +1509,69 @@ func TestIssue32213(t *testing.T) {
tk.MustQuery("select cast(test.t1.c1 as decimal(5, 3)) from test.t1").Check(testkit.Rows("99.999"))
tk.MustQuery("select cast(test.t1.c1 as decimal(6, 3)) from test.t1").Check(testkit.Rows("100.000"))
}

func TestInsertLock(t *testing.T) {
store := testkit.CreateMockStore(t)
tk1 := testkit.NewTestKit(t, store)
tk2 := testkit.NewTestKit(t, store)
tk1.MustExec("use test")
tk2.MustExec("use test")

for _, tt := range []struct {
name string
ddl string
dml string
}{
{
"replace-pk",
"create table t (c int primary key clustered)",
"replace into t values (1)",
},
{
"replace-uk",
"create table t (c int unique key)",
"replace into t values (1)",
},
{
"insert-ingore-pk",
"create table t (c int primary key clustered)",
"insert ignore into t values (1)",
},
{
"insert-ingore-uk",
"create table t (c int unique key)",
"insert ignore into t values (1)",
},
{
"insert-update-pk",
"create table t (c int primary key clustered)",
"insert into t values (1) on duplicate key update c = values(c)",
},
{
"insert-update-uk",
"create table t (c int unique key)",
"insert into t values (1) on duplicate key update c = values(c)",
},
} {
t.Run(tt.name, func(t *testing.T) {
tk1.MustExec("drop table if exists t")
tk1.MustExec(tt.ddl)
tk1.MustExec("insert into t values (1)")
tk1.MustExec("begin")
tk1.MustExec(tt.dml)
done := make(chan struct{})
go func() {
tk2.MustExec("delete from t")
done <- struct{}{}
}()
select {
case <-done:
require.Failf(t, "txn2 is not blocked by %q", tt.dml)
case <-time.After(100 * time.Millisecond):
}
tk1.MustExec("commit")
<-done
tk1.MustQuery("select * from t").Check([][]interface{}{})
})
}
}
36 changes: 20 additions & 16 deletions executor/write.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,22 +137,8 @@ func updateRecord(ctx context.Context, sctx sessionctx.Context, h kv.Handle, old
if sctx.GetSessionVars().ClientCapability&mysql.ClientFoundRows > 0 {
sc.AddAffectedRows(1)
}

physicalID := t.Meta().ID
if pt, ok := t.(table.PartitionedTable); ok {
p, err := pt.GetPartitionByRow(sctx, oldData)
if err != nil {
return false, err
}
physicalID = p.GetPhysicalID()
}

unchangedRowKey := tablecodec.EncodeRowKeyWithHandle(physicalID, h)
txnCtx := sctx.GetSessionVars().TxnCtx
if txnCtx.IsPessimistic {
txnCtx.AddUnchangedRowKey(unchangedRowKey)
}
return false, nil
_, err := appendUnchangedRowForLock(sctx, t, h, oldData)
return false, err
}

// Fill values into on-update-now fields, only if they are really changed.
Expand Down Expand Up @@ -229,6 +215,24 @@ func updateRecord(ctx context.Context, sctx sessionctx.Context, h kv.Handle, old
return true, nil
}

func appendUnchangedRowForLock(sctx sessionctx.Context, t table.Table, h kv.Handle, row []types.Datum) (bool, error) {
txnCtx := sctx.GetSessionVars().TxnCtx
if !txnCtx.IsPessimistic {
return false, nil
}
physicalID := t.Meta().ID
if pt, ok := t.(table.PartitionedTable); ok {
p, err := pt.GetPartitionByRow(sctx, row)
if err != nil {
return false, err
}
physicalID = p.GetPhysicalID()
}
unchangedRowKey := tablecodec.EncodeRowKeyWithHandle(physicalID, h)
txnCtx.AddUnchangedRowKey(unchangedRowKey)
return true, nil
}

func rebaseAutoRandomValue(ctx context.Context, sctx sessionctx.Context, t table.Table, newData *types.Datum, col *table.Column) error {
tableInfo := t.Meta()
if !tableInfo.ContainsAutoRandomBits() {
Expand Down
2 changes: 1 addition & 1 deletion tests/realtikvtest/pessimistictest/pessimistic_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -539,7 +539,7 @@ func TestOptimisticConflicts(t *testing.T) {
tk.MustExec("begin pessimistic")
// This SQL use BatchGet and cache data in the txn snapshot.
// It can be changed to other SQLs that use BatchGet.
tk.MustExec("insert ignore into conflict values (1, 2)")
tk.MustExec("select * from conflict where id in (1, 2, 3)")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this query can be removed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

May I know why the query is unnecessary? IMO, it mean to test the snapshot cache. We cache (1, 2) here and don't want to see the following select-for-update (L547) returns the cached row. Although, it won't read from the cache even when we forget to invalidate the cache because it will read from lock cache firstly now (also, we won't reuse the snapshot with for-update-ts now). I think it's ok to keep the query since the test point still makes sense.


tk2.MustExec("update conflict set c = c - 1")

Expand Down