feat(spanner): Support REPEATABLE_READ for RW transaction

Author: shobhitsg

google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java

@@ -111,6 +111,10 @@ default String getDatabaseRole() {
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    *
    * @return a response with the timestamp at which the write was committed
@@ -186,6 +190,10 @@ CommitResponse writeWithOptions(Iterable<Mutation> mutations, TransactionOption.
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    *
    * @return a response with the timestamp at which the write was committed
@@ -414,6 +422,10 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    */
   TransactionRunner readWriteTransaction(TransactionOption... options);
@@ -454,6 +466,10 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    */
   TransactionManager transactionManager(TransactionOption... options);
@@ -494,6 +510,10 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    */
   AsyncRunner runAsync(TransactionOption... options);
@@ -548,6 +568,10 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *       applied to any other requests on the transaction.
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
+   *   <li>{@link Options#repeatableReadIsolationLevel()}: Request Repeatable Read Isolation Level
+   *       from the backend.
+   *   <li>{@link Options#serializableIsolationLevel()}: Request Serializable Isolation Level from
+   *       the backend.
    * </ul>
    */
   AsyncTransactionManager transactionManagerAsync(TransactionOption... options);

google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java

@@ -21,9 +21,13 @@
 import com.google.spanner.v1.ReadRequest.LockHint;
 import com.google.spanner.v1.ReadRequest.OrderBy;
 import com.google.spanner.v1.RequestOptions.Priority;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import java.io.Serializable;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
 
 /** Specifies options for various spanner operations */
 public final class Options implements Serializable {
@@ -131,7 +135,29 @@ public interface UpdateAdminApiOption extends AdminApiOption {}
   public interface QueryOption {}
 
   /** Marker interface to mark options applicable to write operations */
-  public interface TransactionOption {}
+  public interface TransactionOption {
+    Predicate<TransactionOption> isValidIsolationLevelOption =
+        txnOption ->
+            txnOption instanceof RepeatableReadOption || txnOption instanceof SerializableOption;
+
+    /**
+     * Combines two arrays of TransactionOption, with primaryOptions taking precedence in case of
+     * conflicts. Note that {@link
+     * com.google.cloud.spanner.SpannerOptions.Builder.TransactionOptions} supports only the {@link
+     * IsolationLevel} TransactionOption, meaning spannerOptions will contain a maximum of one
+     * TransactionOption.
+     */
+    static TransactionOption[] combine(
+        TransactionOption[] primaryOptions, TransactionOption[] spannerOptions) {
+      if (spannerOptions == null
+          || Arrays.stream(primaryOptions).anyMatch(isValidIsolationLevelOption)) {
+        return primaryOptions;
+      } else {
+        return Stream.concat(Arrays.stream(primaryOptions), Arrays.stream(spannerOptions))
+            .toArray(TransactionOption[]::new);
+      }
+    }
+  }
 
   /** Marker interface to mark options applicable to update operation. */
   public interface UpdateOption {}
@@ -159,6 +185,22 @@ public static TransactionOption optimisticLock() {
     return OPTIMISTIC_LOCK_OPTION;
   }
 
+  /**
+   * Specifying this instructs the transaction to request Repeatable Read Isolation Level from the
+   * backend.
+   */
+  public static TransactionOption repeatableReadIsolationLevel() {
+    return REPEATABLE_READ_OPTION;
+  }
+
+  /**
+   * Specifying this instructs the transaction to request Serializable Isolation Level from the
+   * backend.
+   */
+  public static TransactionOption serializableIsolationLevel() {
+    return SERIALIZABLE_OPTION;
+  }
+
   /**
    * Specifying this instructs the transaction to be excluded from being recorded in change streams
    * with the DDL option `allow_txn_exclusion=true`. This does not exclude the transaction from
@@ -490,6 +532,26 @@ void appendToOptions(Options options) {
     }
   }
 
+  /** Option to request REPEATABLE READ isolation level for read/write transactions. */
+  static final class RepeatableReadOption extends InternalOption implements TransactionOption {
+    @Override
+    void appendToOptions(Options options) {
+      options.isolationLevel = IsolationLevel.REPEATABLE_READ;
+    }
+  }
+
+  static final RepeatableReadOption REPEATABLE_READ_OPTION = new RepeatableReadOption();
+
+  /** Option to request SERIALIZABLE isolation level for read/write transactions. */
+  static final class SerializableOption extends InternalOption implements TransactionOption {
+    @Override
+    void appendToOptions(Options options) {
+      options.isolationLevel = IsolationLevel.SERIALIZABLE;
+    }
+  }
+
+  static final SerializableOption SERIALIZABLE_OPTION = new SerializableOption();
+
   private boolean withCommitStats;
 
   private Duration maxCommitDelay;
@@ -512,6 +574,7 @@ void appendToOptions(Options options) {
   private RpcOrderBy orderBy;
   private RpcLockHint lockHint;
   private Boolean lastStatement;
+  private IsolationLevel isolationLevel;
 
   // Construction is via factory methods below.
   private Options() {}
@@ -664,6 +727,10 @@ LockHint lockHint() {
     return lockHint == null ? null : lockHint.proto;
   }
 
+  IsolationLevel isolationLevel() {
+    return isolationLevel;
+  }
+
   @Override
   public String toString() {
     StringBuilder b = new StringBuilder();
@@ -726,6 +793,9 @@ public String toString() {
     if (lockHint != null) {
       b.append("lockHint: ").append(lockHint).append(' ');
     }
+    if (isolationLevel != null) {
+      b.append("isolationLevel: ").append(isolationLevel).append(' ');
+    }
     return b.toString();
   }
 
@@ -767,7 +837,8 @@ public boolean equals(Object o) {
         && Objects.equals(directedReadOptions(), that.directedReadOptions())
         && Objects.equals(orderBy(), that.orderBy())
         && Objects.equals(isLastStatement(), that.isLastStatement())
-        && Objects.equals(lockHint(), that.lockHint());
+        && Objects.equals(lockHint(), that.lockHint())
+        && Objects.equals(isolationLevel(), that.isolationLevel());
   }
 
   @Override
@@ -833,6 +904,9 @@ public int hashCode() {
     if (lockHint != null) {
       result = 31 * result + lockHint.hashCode();
     }
+    if (isolationLevel != null) {
+      result = 31 * result + isolationLevel.hashCode();
+    }
     return result;
   }
 

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java

@@ -83,6 +83,9 @@ static TransactionOptions createReadWriteTransactionOptions(
         && previousTransactionId != com.google.protobuf.ByteString.EMPTY) {
       readWrite.setMultiplexedSessionPreviousTransactionId(previousTransactionId);
     }
+    if (options.isolationLevel() != null) {
+      transactionOptions.setIsolationLevel(options.isolationLevel());
+    }
     transactionOptions.setReadWrite(readWrite);
     return transactionOptions.build();
   }
@@ -239,7 +242,10 @@ public CommitResponse writeAtLeastOnceWithOptions(
     setActive(null);
     List<com.google.spanner.v1.Mutation> mutationsProto = new ArrayList<>();
     Mutation.toProtoAndReturnRandomMutation(mutations, mutationsProto);
-    Options options = Options.fromTransactionOptions(transactionOptions);
+    Options options =
+        Options.fromTransactionOptions(
+            TransactionOption.combine(
+                transactionOptions, this.spanner.getOptions().getTransactionOptions()));
     final CommitRequest.Builder requestBuilder =
         CommitRequest.newBuilder()
             .setSession(getName())
@@ -252,6 +258,9 @@ public CommitResponse writeAtLeastOnceWithOptions(
     if (options.withExcludeTxnFromChangeStreams() == Boolean.TRUE) {
       transactionOptionsBuilder.setExcludeTxnFromChangeStreams(true);
     }
+    if (options.isolationLevel() != null) {
+      transactionOptionsBuilder.setIsolationLevel(options.isolationLevel());
+    }
     requestBuilder.setSingleUseTransaction(transactionOptionsBuilder);
 
     if (options.hasMaxCommitDelay()) {
@@ -396,22 +405,37 @@ public ReadOnlyTransaction readOnlyTransaction(TimestampBound bound) {
 
   @Override
   public TransactionRunner readWriteTransaction(TransactionOption... options) {
-    return setActive(new TransactionRunnerImpl(this, options));
+    return setActive(
+        new TransactionRunnerImpl(
+            this,
+            TransactionOption.combine(options, this.spanner.getOptions().getTransactionOptions())));
   }
 
   @Override
   public AsyncRunner runAsync(TransactionOption... options) {
-    return new AsyncRunnerImpl(setActive(new TransactionRunnerImpl(this, options)));
+    return new AsyncRunnerImpl(
+        setActive(
+            new TransactionRunnerImpl(
+                this,
+                TransactionOption.combine(
+                    options, this.spanner.getOptions().getTransactionOptions()))));
   }
 
   @Override
   public TransactionManager transactionManager(TransactionOption... options) {
-    return new TransactionManagerImpl(this, currentSpan, tracer, options);
+    return new TransactionManagerImpl(
+        this,
+        currentSpan,
+        tracer,
+        TransactionOption.combine(options, this.spanner.getOptions().getTransactionOptions()));
   }
 
   @Override
   public AsyncTransactionManagerImpl transactionManagerAsync(TransactionOption... options) {
-    return new AsyncTransactionManagerImpl(this, currentSpan, options);
+    return new AsyncTransactionManagerImpl(
+        this,
+        currentSpan,
+        TransactionOption.combine(options, this.spanner.getOptions().getTransactionOptions()));
   }
 
   @Override