Savepoint Construct Implementation in VTGate

go/mysql/constants.go

@@ -438,6 +438,7 @@ const (
 	ERInvalidOnUpdate              = 1294
 	ERUnknownTimeZone              = 1298
 	ERInvalidCharacterString       = 1300
+	ERSavepointNotExist            = 1305
 	ERIllegalReference             = 1247
 	ERDerivedMustHaveAlias         = 1248
 	ERTableNameNotAllowedHere      = 1250

go/test/endtoend/vtgate/misc_test.go

@@ -608,6 +608,117 @@ func TestUnion(t *testing.T) {
 	assertMatches(t, conn, `(SELECT 1,'a' order by 1) union (SELECT 1,'a' ORDER BY 1)`, `[[INT64(1) VARCHAR("a")]]`)
 }
 
+func TestSavepointInTx(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	exec(t, conn, "savepoint a")
+	exec(t, conn, "start transaction")
+	exec(t, conn, "savepoint b")
+	exec(t, conn, "rollback to b")
+	exec(t, conn, "release savepoint b")
+	exec(t, conn, "savepoint b")
+	exec(t, conn, "insert into t1(id1, id2) values(1,1)") // -80
+	exec(t, conn, "savepoint c")
+	exec(t, conn, "insert into t1(id1, id2) values(4,4)") // 80-
+	exec(t, conn, "savepoint d")
+	exec(t, conn, "insert into t1(id1, id2) values(2,2)") // -80
+	exec(t, conn, "savepoint e")
+
+	// Validate all the data.
+	exec(t, conn, "use `ks:-80`")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)]]`)
+	exec(t, conn, "use `ks:80-`")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(4)]]`)
+	exec(t, conn, "use")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)] [INT64(4)]]`)
+
+	_, err = conn.ExecuteFetch("rollback work to savepoint a", 1000, true)
+	require.Error(t, err)
+
+	exec(t, conn, "release savepoint d")
+
+	_, err = conn.ExecuteFetch("rollback to d", 1000, true)
+	require.Error(t, err)
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)] [INT64(4)]]`)
+
+	exec(t, conn, "rollback to c")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)]]`)
+
+	exec(t, conn, "insert into t1(id1, id2) values(2,2),(3,3),(4,4)")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)] [INT64(3)] [INT64(4)]]`)
+
+	exec(t, conn, "rollback to b")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+
+	exec(t, conn, "commit")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+
+	exec(t, conn, "start transaction")
+
+	exec(t, conn, "insert into t1(id1, id2) values(2,2),(3,3),(4,4)")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(2)] [INT64(3)] [INT64(4)]]`)
+
+	// After previous commit all the savepoints are cleared.
+	_, err = conn.ExecuteFetch("rollback to b", 1000, true)
+	require.Error(t, err)
+
+	exec(t, conn, "rollback")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+}
+
+func TestSavepointOutsideTx(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	exec(t, conn, "savepoint a")
+	exec(t, conn, "savepoint b")
+
+	_, err = conn.ExecuteFetch("rollback to b", 1, true)
+	require.Error(t, err)
+	_, err = conn.ExecuteFetch("release savepoint a", 1, true)
+	require.Error(t, err)
+}
+
+func TestSavepointAdditionalCase(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	exec(t, conn, "start transaction")
+	exec(t, conn, "savepoint a")
+	exec(t, conn, "insert into t1(id1, id2) values(1,1)")             // -80
+	exec(t, conn, "insert into t1(id1, id2) values(2,2),(3,3),(4,4)") // -80 & 80-
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)] [INT64(3)] [INT64(4)]]`)
+
+	exec(t, conn, "rollback to a")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+
+	exec(t, conn, "commit")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+
+	exec(t, conn, "start transaction")
+	exec(t, conn, "insert into t1(id1, id2) values(1,1)") // -80
+	exec(t, conn, "savepoint a")
+	exec(t, conn, "insert into t1(id1, id2) values(2,2),(3,3)") // -80
+	exec(t, conn, "insert into t1(id1, id2) values(4,4)")       // 80-
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)] [INT64(2)] [INT64(3)] [INT64(4)]]`)
+
+	exec(t, conn, "rollback to a")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[[INT64(1)]]`)
+
+	exec(t, conn, "rollback")
+	assertMatches(t, conn, "select id1 from t1 order by id1", `[]`)
+}
+
 func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
 	t.Helper()
 	qr := exec(t, conn, query)