Add transaction-level isolation level setting (#206)

Enables optionally setting the isolation level at a transaction-level to either `SERIALIZABLE` or `REPEATABLE READ`. If the isolation level is not set, Spanner will default to `SERIALIZABLE` isolation.

Author: skuruppu

README.md

@@ -237,7 +237,7 @@ and `{}` for a mutually exclusive keyword.
 | Show Query Result Shape | `DESCRIBE SELECT ...;` | |
 | Show DML Result Shape | `DESCRIBE {INSERT\|UPDATE\|DELETE} ... THEN RETURN ...;` | |
 | Start a new query optimizer statistics package construction | `ANALYZE;` | |
-| Start Read-Write Transaction | `BEGIN [RW] [PRIORITY {HIGH\|MEDIUM\|LOW}] [TAG <tag>];` | See [Request Priority](#request-priority) for details on the priority. The tag you set is used as both transaction tag and request tag. See also [Transaction Tags and Request Tags](#transaction-tags-and-request-tags).|
+| Start Read-Write Transaction | `BEGIN [RW] [ISOLATION LEVEL {SERIALIZABLE\|REPEATABLE READ}] [PRIORITY {HIGH\|MEDIUM\|LOW}] [TAG <tag>];` | See [Isolation Level](#isolation-level) for details on the isolation level. See [Request Priority](#request-priority) for details on the priority. The tag you set is used as both transaction tag and request tag. See also [Transaction Tags and Request Tags](#transaction-tags-and-request-tags).|
 | Commit Read-Write Transaction | `COMMIT;` | |
 | Rollback Read-Write Transaction | `ROLLBACK;` | |
 | Start Read-Only Transaction | `BEGIN RO [{<seconds>\|<RFC3339-formatted time>}] [PRIORITY {HIGH\|MEDIUM\|LOW}] [TAG <tag>];` | `<seconds>` and `<RFC3339-formatted time>` is used for stale read. See [Request Priority](#request-priority) for details on the priority. The tag you set is used as request tag. See also [Transaction Tags and Request Tags](#transaction-tags-and-request-tags).|
@@ -300,14 +300,32 @@ prompt = "[\\p:\\i:\\d]\\t> "
 3. `.spanner_cli.cnf` in current directory
 4. `.spanner_cli.cnf` in home directory(lowest)
 
+## Isolation Level
+
+You can set the isolation level at a transaction level in read-write transactions. By default `SERIALIZABLE` isolation is used for every request.
+
+To set the isolation level for a transaction, you can use the `ISOLATION LEVEL {SERIALIZABLE|REPEATABLE READ}` keyword.
+
+Here are some examples for transaction-level isolation.
+
+```
+# Read-write transaction with serializable isolation
+BEGIN RW ISOLATION LEVEL SERIALIZABLE
+
+# Read-write transaction with repeatable read isolation
+BEGIN RW ISOLATION LEVEL REPEATABLE READ
+```
+
+Note that the transaction-level isolation level cannot be set on read-only transactions.
+
 ## Request Priority
 
 You can set [request priority](https://cloud.google.com/spanner/docs/reference/rest/v1/RequestOptions#Priority) for command level or transaction level.
 By default `MEDIUM` priority is used for every request.
 
 To set a priority for command line level, you can use `--priority={HIGH|MEDIUM|LOW}` command line option.
 
-To set a priority for transaction level, you can use `PRIORITY {HIGH|MEDIUM|LOW}` keyword.
+To set a priority for transaction level, you can use the `PRIORITY {HIGH|MEDIUM|LOW}` keyword.
 
 Here are some examples for transaction-level priority.
 

session.go

@@ -119,7 +119,7 @@ func (s *Session) InReadOnlyTransaction() bool {
 }
 
 // BeginReadWriteTransaction starts read-write transaction.
-func (s *Session) BeginReadWriteTransaction(ctx context.Context, priority pb.RequestOptions_Priority, tag string) error {
+func (s *Session) BeginReadWriteTransaction(ctx context.Context, isolation_level pb.TransactionOptions_IsolationLevel, priority pb.RequestOptions_Priority, tag string) error {
 	if s.InReadWriteTransaction() {
 		return errors.New("read-write transaction is already running")
 	}
@@ -132,6 +132,7 @@ func (s *Session) BeginReadWriteTransaction(ctx context.Context, priority pb.Req
 	opts := spanner.TransactionOptions{
 		CommitOptions:  spanner.CommitOptions{ReturnCommitStats: true},
 		CommitPriority: priority,
+		IsolationLevel: isolation_level,
 		TransactionTag: tag,
 	}
 	txn, err := spanner.NewReadWriteStmtBasedTransactionWithOptions(ctx, s.client, opts)

session_test.go

@@ -72,7 +72,7 @@ func TestRequestPriority(t *testing.T) {
 			}
 
 			// Read-Write Transaction.
-			if err := session.BeginReadWriteTransaction(ctx, test.transactionPriority, ""); err != nil {
+			if err := session.BeginReadWriteTransaction(ctx, pb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED, test.transactionPriority, ""); err != nil {
 				t.Fatalf("failed to begin read write transaction: %v", err)
 			}
 			iter, _ := session.RunQuery(ctx, spanner.NewStatement("SELECT * FROM t1"))
@@ -191,6 +191,81 @@ func TestParseDirectedReadOption(t *testing.T) {
 	}
 }
 
+func TestIsolationLevel(t *testing.T) {
+	server := setupTestServer(t)
+
+	var recorder requestRecorder
+	unaryInterceptor, streamInterceptor := recordRequestsInterceptors(&recorder)
+	opts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithUnaryInterceptor(unaryInterceptor),
+		grpc.WithStreamInterceptor(streamInterceptor),
+	}
+	ctx := context.Background()
+	conn, err := grpc.DialContext(ctx, server.Addr, opts...)
+	if err != nil {
+		t.Fatalf("failed to dial: %v", err)
+	}
+
+	for _, test := range []struct {
+		desc           string
+		isolationLevel pb.TransactionOptions_IsolationLevel
+		want           pb.TransactionOptions_IsolationLevel
+	}{
+		{
+			desc:           "use default unspecified isolation level",
+			isolationLevel: pb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED,
+			want:           pb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED,
+		},
+		{
+			desc:           "use SERIALIZABLE isolation level",
+			isolationLevel: pb.TransactionOptions_SERIALIZABLE,
+			want:           pb.TransactionOptions_SERIALIZABLE,
+		},
+		{
+			desc:           "use REPEATABLE READ isolation level",
+			isolationLevel: pb.TransactionOptions_REPEATABLE_READ,
+			want:           pb.TransactionOptions_REPEATABLE_READ,
+		},
+	} {
+		t.Run(test.desc, func(t *testing.T) {
+			defer recorder.flush()
+
+			session, err := NewSession("project", "instance", "database", pb.RequestOptions_PRIORITY_UNSPECIFIED, "role", nil, nil, option.WithGRPCConn(conn))
+			if err != nil {
+				t.Fatalf("failed to create spanner-cli session: %v", err)
+			}
+
+			// Read-Write Transaction.
+			if err := session.BeginReadWriteTransaction(ctx, test.isolationLevel, pb.RequestOptions_PRIORITY_UNSPECIFIED, ""); err != nil {
+				t.Fatalf("failed to begin read write transaction: %v", err)
+			}
+			iter, _ := session.RunQuery(ctx, spanner.NewStatement("SELECT * FROM t1"))
+			if err := iter.Do(func(r *spanner.Row) error {
+				return nil
+			}); err != nil {
+				t.Fatalf("failed to run query: %v", err)
+			}
+			if _, _, _, _, err := session.RunUpdate(ctx, spanner.NewStatement("DELETE FROM t1 WHERE Id = 1"), true); err != nil {
+				t.Fatalf("failed to run update: %v", err)
+			}
+			if _, err := session.CommitReadWriteTransaction(ctx); err != nil {
+				t.Fatalf("failed to commit: %v", err)
+			}
+
+			// Check request priority.
+			for _, r := range recorder.requests {
+				switch v := r.(type) {
+				case *pb.BeginTransactionRequest:
+					if got := v.GetOptions().GetIsolationLevel(); got != test.want {
+						t.Errorf("isolation level mismatch: got = %v, want = %v", got, test.want)
+					}
+				}
+			}
+		})
+	}
+}
+
 func setupTestServer(t *testing.T) *spannertest.Server {
 	server, err := spannertest.NewServer("localhost:0")
 	if err != nil {

statement.go

@@ -110,7 +110,7 @@ var (
 	pdmlRe = regexp.MustCompile(`(?is)^PARTITIONED\s+((?:INSERT|UPDATE|DELETE)\s+.+$)`)
 
 	// Transaction
-	beginRwRe  = regexp.MustCompile(`(?is)^BEGIN(?:\s+RW)?(?:\s+PRIORITY\s+(HIGH|MEDIUM|LOW))?(?:\s+TAG\s+(.+))?$`)
+	beginRwRe  = regexp.MustCompile(`(?is)^BEGIN(?:\s+RW)?(?:\s+ISOLATION LEVEL\s+(SERIALIZABLE|REPEATABLE READ))?(?:\s+PRIORITY\s+(HIGH|MEDIUM|LOW))?(?:\s+TAG\s+(.+))?$`)
 	beginRoRe  = regexp.MustCompile(`(?is)^BEGIN\s+RO(?:\s+([^\s]+))?(?:\s+PRIORITY\s+(HIGH|MEDIUM|LOW))?(?:\s+TAG\s+(.+))?$`)
 	commitRe   = regexp.MustCompile(`(?is)^COMMIT$`)
 	rollbackRe = regexp.MustCompile(`(?is)^ROLLBACK$`)
@@ -991,24 +991,33 @@ func runInNewOrExistRwTxForExplain(ctx context.Context, session *Session, f func
 }
 
 type BeginRwStatement struct {
-	Priority pb.RequestOptions_Priority
-	Tag      string
+	IsolationLevel pb.TransactionOptions_IsolationLevel
+	Priority       pb.RequestOptions_Priority
+	Tag            string
 }
 
 func newBeginRwStatement(input string) (*BeginRwStatement, error) {
 	matched := beginRwRe.FindStringSubmatch(input)
 	stmt := &BeginRwStatement{}
 
 	if matched[1] != "" {
-		priority, err := parsePriority(matched[1])
+		isolationLevel, err := parseIsolationLevel(matched[1])
 		if err != nil {
 			return nil, err
 		}
-		stmt.Priority = priority
+		stmt.IsolationLevel = isolationLevel
 	}
 
 	if matched[2] != "" {
-		stmt.Tag = matched[2]
+		priority, err := parsePriority(matched[2])
+		if err != nil {
+			return nil, err
+		}
+		stmt.Priority = priority
+	}
+
+	if matched[3] != "" {
+		stmt.Tag = matched[3]
 	}
 
 	return stmt, nil
@@ -1022,7 +1031,7 @@ func (s *BeginRwStatement) Execute(ctx context.Context, session *Session) (*Resu
 		return nil, errors.New("you're in read-only transaction. Please finish the transaction by 'CLOSE;'")
 	}
 
-	if err := session.BeginReadWriteTransaction(ctx, s.Priority, s.Tag); err != nil {
+	if err := session.BeginReadWriteTransaction(ctx, s.IsolationLevel, s.Priority, s.Tag); err != nil {
 		return nil, err
 	}
 
@@ -1190,3 +1199,14 @@ func parsePriority(priority string) (pb.RequestOptions_Priority, error) {
 		return pb.RequestOptions_PRIORITY_UNSPECIFIED, fmt.Errorf("invalid priority: %q", priority)
 	}
 }
+
+func parseIsolationLevel(isolationLevel string) (pb.TransactionOptions_IsolationLevel, error) {
+	switch strings.ToUpper(isolationLevel) {
+	case "SERIALIZABLE":
+		return pb.TransactionOptions_SERIALIZABLE, nil
+	case "REPEATABLE READ":
+		return pb.TransactionOptions_REPEATABLE_READ, nil
+	default:
+		return pb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED, fmt.Errorf("invalid isolation level: %q", isolationLevel)
+	}
+}

statement_test.go

@@ -321,6 +321,45 @@ func TestBuildStatement(t *testing.T) {
 				Tag: "app=spanner-cli env=test",
 			},
 		},
+		{
+			desc:  "BEGIN RW statement with ISOLATION LEVEL",
+			input: "BEGIN RW ISOLATION LEVEL SERIALIZABLE",
+			want: &BeginRwStatement{
+				IsolationLevel: pb.TransactionOptions_SERIALIZABLE,
+			},
+		},
+		{
+			desc:  "BEGIN RW statement with ISOLATION LEVEL",
+			input: "BEGIN RW ISOLATION LEVEL REPEATABLE READ",
+			want: &BeginRwStatement{
+				IsolationLevel: pb.TransactionOptions_REPEATABLE_READ,
+			},
+		},
+		{
+			desc:  "BEGIN RW statement with ISOLATION LEVEL and PRIORITY",
+			input: "BEGIN RW ISOLATION LEVEL SERIALIZABLE PRIORITY MEDIUM",
+			want: &BeginRwStatement{
+				IsolationLevel: pb.TransactionOptions_SERIALIZABLE,
+				Priority:       pb.RequestOptions_PRIORITY_MEDIUM,
+			},
+		},
+		{
+			desc:  "BEGIN RW statement with ISOLATION LEVEL and TAG",
+			input: "BEGIN RW ISOLATION LEVEL SERIALIZABLE TAG app=spanner-cli",
+			want: &BeginRwStatement{
+				IsolationLevel: pb.TransactionOptions_SERIALIZABLE,
+				Tag:            "app=spanner-cli",
+			},
+		},
+		{
+			desc:  "BEGIN RW statement with ISOLATION LEVEL, PRIORITY and TAG",
+			input: "BEGIN RW ISOLATION LEVEL REPEATABLE READ PRIORITY MEDIUM TAG app=spanner-cli",
+			want: &BeginRwStatement{
+				IsolationLevel: pb.TransactionOptions_REPEATABLE_READ,
+				Priority:       pb.RequestOptions_PRIORITY_MEDIUM,
+				Tag:            "app=spanner-cli",
+			},
+		},
 		{
 			desc:  "BEGIN PRIORITY statement with TAG whitespace",
 			input: "BEGIN PRIORITY MEDIUM TAG app=spanner-cli env=test",