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

Add savepoint support to atomic distributed transaction #16863

Merged
merged 8 commits into from
Oct 10, 2024
Merged

Add savepoint support to atomic distributed transaction #16863

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6ba31ad
allow savepoint queries to be stored in redo log statement, allow com…
harshit-gangal Sep 30, 2024
6fb1004
test: savepoint reject test removed
harshit-gangal Sep 30, 2024
4c705eb
savepoint based action on redo statements
harshit-gangal Oct 1, 2024
e8ada40
execute savepoints after the begin method is complete in the tabletse…
harshit-gangal Oct 7, 2024
69878ba
test: update test expectation with savepoint
harshit-gangal Oct 8, 2024
35cf752
addressed review comments
harshit-gangal Oct 8, 2024
a8eab2c
test: add savepoints to the fuzzer
GuptaManan100 Oct 10, 2024
bc836c4
fix: test setup to run reshard
harshit-gangal Oct 10, 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
37 changes: 36 additions & 1 deletion 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 @@ -24,6 +24,7 @@ import (
"fmt"
"io"
"os"
"strings"
"sync"
"testing"
"time"
Expand Down Expand Up @@ -123,6 +124,7 @@ func cleanup(t *testing.T) {
cluster.PanicHandler(t)
utils.ClearOutTable(t, vtParams, "twopc_user")
utils.ClearOutTable(t, vtParams, "twopc_t1")
sm.reset()
}

type extractInterestingValues func(dtidMap map[string]string, vals []sqltypes.Value) []sqltypes.Value
Expand All @@ -147,7 +149,8 @@ var tables = map[string]extractInterestingValues{
},
"ks.redo_statement": func(dtidMap map[string]string, vals []sqltypes.Value) (out []sqltypes.Value) {
dtid := getDTID(dtidMap, vals[0].ToString())
out = append([]sqltypes.Value{sqltypes.NewVarChar(dtid)}, vals[1:]...)
stmt := getStatement(vals[2].ToString())
out = append([]sqltypes.Value{sqltypes.NewVarChar(dtid)}, vals[1], sqltypes.TestValue(sqltypes.Blob, stmt))
return
},
"ks.twopc_user": func(_ map[string]string, vals []sqltypes.Value) []sqltypes.Value { return vals },
Expand All @@ -167,6 +170,28 @@ func getDTID(dtidMap map[string]string, dtKey string) string {
return dtid
}

func getStatement(stmt string) string {
var sKey string
var prefix string
switch {
case strings.HasPrefix(stmt, "savepoint"):
prefix = "savepoint-"
sKey = stmt[9:]
case strings.HasPrefix(stmt, "rollback to"):
prefix = "rollback-"
sKey = stmt[11:]
default:
return stmt
}

sid, exists := sm.stmt[sKey]
if !exists {
sid = fmt.Sprintf("%d", len(sm.stmt)+1)
sm.stmt[sKey] = sid
}
return prefix + sid
}

func runVStream(t *testing.T, ctx context.Context, ch chan *binlogdatapb.VEvent, vtgateConn *vtgateconn.VTGateConn) {
vgtid := &binlogdatapb.VGtid{
ShardGtids: []*binlogdatapb.ShardGtid{
Expand Down Expand Up @@ -272,3 +297,13 @@ func prettyPrint(v interface{}) string {
}
return string(b)
}

type stmtMapper struct {
stmt map[string]string
}

var sm = &stmtMapper{stmt: make(map[string]string)}

func (sm *stmtMapper) reset() {
sm.stmt = make(map[string]string)
}
Loading
Loading