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

Consistent lookup vindex tests for atomic distributed transactions #17393

Merged
merged 6 commits into from
Dec 18, 2024
+275 −99
Merged

Consistent lookup vindex tests for atomic distributed transactions #17393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
dd6ef5b
feat: fix lookup vindex schema and add consistent lookup vindex
GuptaManan100 Dec 9, 2024
2b4d1f6
feat: add consistent lookup vindex tests and make them work
GuptaManan100 Dec 15, 2024
ed14de1
Merge remote-tracking branch 'upstream/main' into consistent-lookup-v…
GuptaManan100 Dec 15, 2024
9c750d0
test: remove test that fails
GuptaManan100 Dec 16, 2024
b780dd6
feat: add remaining tests
GuptaManan100 Dec 16, 2024
6055df4
test: fix comments in the test
GuptaManan100 Dec 18, 2024
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
3 changes: 3 additions & 0 deletions go/test/endtoend/transaction/twopc/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,9 @@ func cleanup(t *testing.T) {
twopcutil.ClearOutTable(t, vtParams, "twopc_lookup")
twopcutil.ClearOutTable(t, vtParams, "lookup_unique")
twopcutil.ClearOutTable(t, vtParams, "lookup")
twopcutil.ClearOutTable(t, vtParams, "twopc_consistent_lookup")
twopcutil.ClearOutTable(t, vtParams, "consistent_lookup_unique")
twopcutil.ClearOutTable(t, vtParams, "consistent_lookup")
sm.reset()
}

Expand Down
29 changes: 26 additions & 3 deletions go/test/endtoend/transaction/twopc/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,38 @@ create table twopc_lookup

create table lookup
(
col varchar(128),
col bigint,
id bigint,
keyspace_id varbinary(100),
primary key (id)
primary key (col, id)
) Engine = InnoDB;

create table lookup_unique
(
col_unique varchar(128),
col_unique bigint,
keyspace_id varbinary(100),
primary key (col_unique)
) Engine = InnoDB;

create table twopc_consistent_lookup
(
id bigint,
col bigint,
col_unique bigint,
primary key (id)
) Engine=InnoDB;

create table consistent_lookup
(
col bigint,
id bigint,
keyspace_id varbinary(100),
primary key (col, id)
) Engine = InnoDB;

create table consistent_lookup_unique
(
col_unique bigint,
keyspace_id varbinary(100),
primary key (col_unique)
) Engine = InnoDB;
183 changes: 167 additions & 16 deletions go/test/endtoend/transaction/twopc/twopc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -563,7 +563,11 @@ func compareMaps(t *testing.T, expected, actual map[string][]string, flexibleExp
// TestDTResolveAfterMMCommit tests that transaction is committed on recovery
// failure after MM commit.
func TestDTResolveAfterMMCommit(t *testing.T) {
defer cleanup(t)
initconn, closer := start(t)
defer closer()

// Do an insertion into a table that has a consistent lookup vindex.
utils.Exec(t, initconn, "insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)")

vtgateConn, err := cluster.DialVTGate(context.Background(), t.Name(), vtgateGrpcAddress, "dt_user", "")
require.NoError(t, err)
Expand All @@ -589,6 +593,10 @@ func TestDTResolveAfterMMCommit(t *testing.T) {
require.NoError(t, err)
_, err = conn.Execute(qCtx, "insert into twopc_user(id, name) values(10,'apa')", nil)
require.NoError(t, err)
// Also do an update to a table that has a consistent lookup vindex.
// We expect to see only the pre-session changes in the logs.
_, err = conn.Execute(qCtx, "update twopc_consistent_lookup set col = 22 where id = 4", nil)
require.NoError(t, err)

// The caller ID is used to simulate the failure at the desired point.
newCtx := callerid.NewContext(qCtx, callerid.NewEffectiveCallerID("MMCommitted_FailNow", "", ""), nil)
Expand Down Expand Up @@ -625,7 +633,9 @@ func TestDTResolveAfterMMCommit(t *testing.T) {
},
"ks.redo_statement:-40": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"insert into twopc_user(id, `name`) values (10, 'apa')\")]",
"insert:[VARCHAR(\"dtid-1\") INT64(2) BLOB(\"update twopc_consistent_lookup set col = 22 where id = 4 limit 10001 /* INT64 */\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"insert into twopc_user(id, `name`) values (10, 'apa')\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(2) BLOB(\"update twopc_consistent_lookup set col = 22 where id = 4 limit 10001 /* INT64 */\")]",
},
"ks.redo_statement:40-80": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"insert into twopc_user(id, `name`) values (8, 'bar')\")]",
Expand All @@ -641,6 +651,12 @@ func TestDTResolveAfterMMCommit(t *testing.T) {
`insert:[INT64(7) VARCHAR("foo")]`,
`insert:[INT64(9) VARCHAR("baz")]`,
},
"ks.consistent_lookup:-40": {
"insert:[INT64(22) INT64(4) VARBINARY(\" \\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.twopc_consistent_lookup:-40": {
"update:[INT64(4) INT64(22) INT64(6)]",
},
}
assert.Equal(t, expectations, logTable,
"mismatch expected: \n got: %s, want: %s", prettyPrint(logTable), prettyPrint(expectations))
Expand All @@ -649,7 +665,11 @@ func TestDTResolveAfterMMCommit(t *testing.T) {
// TestDTResolveAfterRMPrepare tests that transaction is rolled back on recovery
// failure after RM prepare and before MM commit.
func TestDTResolveAfterRMPrepare(t *testing.T) {
defer cleanup(t)
initconn, closer := start(t)
defer closer()

// Do an insertion into a table that has a consistent lookup vindex.
utils.Exec(t, initconn, "insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)")

vtgateConn, err := cluster.DialVTGate(context.Background(), t.Name(), vtgateGrpcAddress, "dt_user", "")
require.NoError(t, err)
Expand All @@ -671,6 +691,10 @@ func TestDTResolveAfterRMPrepare(t *testing.T) {
require.NoError(t, err)
_, err = conn.Execute(qCtx, "insert into twopc_user(id, name) values(8,'bar')", nil)
require.NoError(t, err)
// Also do an update to a table that has a consistent lookup vindex.
// We expect to see only the pre-session changes in the logs.
_, err = conn.Execute(qCtx, "update twopc_consistent_lookup set col = 22 where id = 4", nil)
require.NoError(t, err)

// The caller ID is used to simulate the failure at the desired point.
newCtx := callerid.NewContext(qCtx, callerid.NewEffectiveCallerID("RMPrepared_FailNow", "", ""), nil)
Expand All @@ -693,16 +717,29 @@ func TestDTResolveAfterRMPrepare(t *testing.T) {
},
"ks.dt_participant:80-": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) VARCHAR(\"ks\") VARCHAR(\"40-80\")]",
"insert:[VARCHAR(\"dtid-1\") INT64(2) VARCHAR(\"ks\") VARCHAR(\"-40\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) VARCHAR(\"ks\") VARCHAR(\"40-80\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(2) VARCHAR(\"ks\") VARCHAR(\"-40\")]",
},
"ks.redo_state:40-80": {
"insert:[VARCHAR(\"dtid-1\") VARCHAR(\"PREPARE\")]",
"delete:[VARCHAR(\"dtid-1\") VARCHAR(\"PREPARE\")]",
},
"ks.redo_state:-40": {
"insert:[VARCHAR(\"dtid-1\") VARCHAR(\"PREPARE\")]",
"delete:[VARCHAR(\"dtid-1\") VARCHAR(\"PREPARE\")]",
},
"ks.redo_statement:40-80": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"insert into twopc_user(id, `name`) values (8, 'bar')\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"insert into twopc_user(id, `name`) values (8, 'bar')\")]",
},
"ks.redo_statement:-40": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"update twopc_consistent_lookup set col = 22 where id = 4 limit 10001 /* INT64 */\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"update twopc_consistent_lookup set col = 22 where id = 4 limit 10001 /* INT64 */\")]",
},
"ks.consistent_lookup:-40": {
"insert:[INT64(22) INT64(4) VARBINARY(\" \\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
}
assert.Equal(t, expectations, logTable,
"mismatch expected: \n got: %s, want: %s", prettyPrint(logTable), prettyPrint(expectations))
Expand Down Expand Up @@ -1491,8 +1528,8 @@ func TestVindexes(t *testing.T) {
"update:[INT64(6) INT64(9) INT64(9)]",
},
"ks.lookup:80-": {
"delete:[VARCHAR(\"4\") INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[VARCHAR(\"9\") INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(9) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
Expand All @@ -1519,8 +1556,8 @@ func TestVindexes(t *testing.T) {
"update:[INT64(6) INT64(4) INT64(20)]",
},
"ks.lookup_unique:80-": {
"delete:[VARCHAR(\"9\") VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[VARCHAR(\"20\") VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(9) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(20) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
Expand All @@ -1547,10 +1584,10 @@ func TestVindexes(t *testing.T) {
"delete:[INT64(6) INT64(4) INT64(9)]",
},
"ks.lookup_unique:80-": {
"delete:[VARCHAR(\"9\") VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(9) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.lookup:80-": {
"delete:[VARCHAR(\"4\") INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
Expand All @@ -1572,10 +1609,10 @@ func TestVindexes(t *testing.T) {
"delete:[VARCHAR(\"dtid-3\") INT64(1) BLOB(\"insert into lookup(col, id, keyspace_id) values (4, 20, _binary'(\\\\0\\\\0\\\\0\\\\0\\\\0\\\\0\\\\0')\")]",
},
"ks.lookup:80-": {
"insert:[VARCHAR(\"4\") INT64(20) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(4) INT64(20) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.lookup_unique:-40": {
"insert:[VARCHAR(\"22\") VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(22) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.twopc_lookup:-40": {
"insert:[INT64(20) INT64(4) INT64(22)]",
Expand Down Expand Up @@ -1625,16 +1662,130 @@ func TestVindexes(t *testing.T) {
"delete:[INT64(9) INT64(4) INT64(4)]",
},
"ks.lookup_unique:-40": {
"insert:[VARCHAR(\"22\") VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(22) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.lookup_unique:80-": {
"delete:[VARCHAR(\"4\") VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.lookup:80-": {
"insert:[VARCHAR(\"4\") INT64(20) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[VARCHAR(\"4\") INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[VARCHAR(\"9\") INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[VARCHAR(\"4\") INT64(9) VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(4) INT64(20) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(9) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(9) VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
{
name: "Consistent Lookup Single Update",
initQueries: []string{
"insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(6, 4, 9)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(9, 4, 4)",
},
testQueries: []string{
"begin",
"update twopc_consistent_lookup set col = 9 where col_unique = 9",
"commit",
},
logExpected: map[string][]string{
"ks.twopc_consistent_lookup:40-80": {
"update:[INT64(6) INT64(9) INT64(9)]",
},
"ks.consistent_lookup:80-": {
"insert:[INT64(9) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
{
name: "Consistent Lookup-Unique Single Update",
initQueries: []string{
"insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(6, 4, 9)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(9, 4, 4)",
},
testQueries: []string{
"begin",
"update twopc_consistent_lookup set col_unique = 20 where col_unique = 9",
"commit",
},
logExpected: map[string][]string{
"ks.twopc_consistent_lookup:40-80": {
"update:[INT64(6) INT64(4) INT64(20)]",
},
"ks.consistent_lookup_unique:80-": {
"insert:[INT64(20) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(9) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
{
name: "Consistent Lookup And Consistent Lookup-Unique Single Delete",
initQueries: []string{
"insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(6, 4, 9)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(9, 4, 4)",
},
testQueries: []string{
"begin",
"delete from twopc_consistent_lookup where col_unique = 9",
"commit",
},
logExpected: map[string][]string{
"ks.twopc_consistent_lookup:40-80": {
"delete:[INT64(6) INT64(4) INT64(9)]",
},
"ks.consistent_lookup_unique:80-": {
"delete:[INT64(9) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.consistent_lookup:80-": {
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
{
name: "Consistent Lookup And Consistent Lookup-Unique Mix",
initQueries: []string{
"insert into twopc_consistent_lookup(id, col, col_unique) values(4, 4, 6)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(6, 4, 9)",
"insert into twopc_consistent_lookup(id, col, col_unique) values(9, 4, 4)",
},
testQueries: []string{
"begin",
"insert into twopc_consistent_lookup(id, col, col_unique) values(20, 4, 22)",
"update twopc_consistent_lookup set col = 9 where col_unique = 9",
"delete from twopc_consistent_lookup where id = 9",
"commit",
},
logExpected: map[string][]string{
"ks.redo_statement:80-": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"delete from twopc_consistent_lookup where id = 9 limit 10001 /* INT64 */\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"delete from twopc_consistent_lookup where id = 9 limit 10001 /* INT64 */\")]",
},
"ks.redo_statement:40-80": {
"insert:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"update twopc_consistent_lookup set col = 9 where col_unique = 9 limit 10001 /* INT64 */\")]",
"delete:[VARCHAR(\"dtid-1\") INT64(1) BLOB(\"update twopc_consistent_lookup set col = 9 where col_unique = 9 limit 10001 /* INT64 */\")]",
},
"ks.twopc_consistent_lookup:-40": {
"insert:[INT64(20) INT64(4) INT64(22)]",
},
"ks.twopc_consistent_lookup:40-80": {
"update:[INT64(6) INT64(9) INT64(9)]",
},
"ks.twopc_consistent_lookup:80-": {
"delete:[INT64(9) INT64(4) INT64(4)]",
},
"ks.consistent_lookup_unique:-40": {
"insert:[INT64(22) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.consistent_lookup_unique:80-": {
"delete:[INT64(4) VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
"ks.consistent_lookup:80-": {
"insert:[INT64(4) INT64(20) VARBINARY(\"(\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"insert:[INT64(9) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(6) VARBINARY(\"`\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
"delete:[INT64(4) INT64(9) VARBINARY(\"\\x90\\x00\\x00\\x00\\x00\\x00\\x00\\x00\")]",
},
},
},
Expand Down
Loading
Loading