feat: add support for Concurrency Control

Author: shobhitsg

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

@@ -24,6 +24,7 @@
 import com.google.cloud.spanner.Statement.StatementFactory;
 import com.google.spanner.v1.BatchWriteResponse;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 
 /**
  * Interface for all the APIs that are used to read/write data into a Cloud Spanner database. An
@@ -417,6 +418,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   TransactionRunner readWriteTransaction(TransactionOption... options);
@@ -458,6 +460,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   TransactionManager transactionManager(TransactionOption... options);
@@ -499,6 +502,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   AsyncRunner runAsync(TransactionOption... options);
@@ -554,6 +558,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   AsyncTransactionManager transactionManagerAsync(TransactionOption... options);

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

@@ -22,6 +22,7 @@
 import com.google.spanner.v1.ReadRequest.OrderBy;
 import com.google.spanner.v1.RequestOptions.Priority;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.Objects;
@@ -155,9 +156,36 @@ public static TransactionOption commitStats() {
    * process in the commit phase (when any needed locks are acquired). The validation process
    * succeeds only if there are no conflicting committed transactions (that committed mutations to
    * the read data at a commit timestamp after the read timestamp).
+   *
+   * @deprecated Use {@link Options#readLockMode(ReadLockMode)} instead.
    */
+  @Deprecated
   public static TransactionOption optimisticLock() {
-    return OPTIMISTIC_LOCK_OPTION;
+    return Options.readLockMode(ReadLockMode.OPTIMISTIC);
+  }
+
+  /**
+   * Specifying this instructs the transaction to request {@link ReadLockMode} from the backend.
+   *
+   * <p>This option controls the locking behavior for read operations and queries within a
+   * read-write transaction. It works in conjunction with the transaction's {@link IsolationLevel}.
+   *
+   * <ul>
+   *   <li>{@link ReadLockMode#OPTIMISTIC}: Reads and queries do not acquire locks during the
+   *       transaction's execution phase. Instead, Spanner validates at commit time that the data
+   *       read has not been modified by other transactions since the read occurred. This mode can
+   *       reduce lock contention and improve read performance but may result in more transaction
+   *       aborts under high write contention on the same data.
+   *   <li>{@link ReadLockMode#PESSIMISTIC}: The transaction acquires locks on data as it's
+   *       accessed. For example, Data Manipulation Language (DML) statements and {@code SELECT ...
+   *       FOR UPDATE} queries will take locks. This approach blocks other transactions from
+   *       modifying the locked data until the current transaction completes, which can prevent
+   *       conflicts and reduce commit-time aborts, but may increase lock wait times and overall
+   *       contention.
+   * </ul>
+   */
+  public static TransactionOption readLockMode(ReadLockMode readLockMode) {
+    return new ReadLockModeOption(readLockMode);
   }
 
   /**
@@ -367,16 +395,6 @@ void appendToOptions(Options options) {
     }
   }
 
-  /** Option to request Optimistic Concurrency Control for read/write transactions. */
-  static final class OptimisticLockOption extends InternalOption implements TransactionOption {
-    @Override
-    void appendToOptions(Options options) {
-      options.withOptimisticLock = true;
-    }
-  }
-
-  static final OptimisticLockOption OPTIMISTIC_LOCK_OPTION = new OptimisticLockOption();
-
   /** Option to request the transaction to be excluded from change streams. */
   static final class ExcludeTxnFromChangeStreamsOption extends InternalOption
       implements UpdateTransactionOption {
@@ -516,6 +534,20 @@ void appendToOptions(Options options) {
     }
   }
 
+  /** Option to set read lock mode for read/write transactions. */
+  static final class ReadLockModeOption extends InternalOption implements TransactionOption {
+    private final ReadLockMode readLockMode;
+
+    public ReadLockModeOption(ReadLockMode readLockMode) {
+      this.readLockMode = readLockMode;
+    }
+
+    @Override
+    void appendToOptions(Options options) {
+      options.readLockMode = readLockMode;
+    }
+  }
+
   private boolean withCommitStats;
 
   private Duration maxCommitDelay;
@@ -530,7 +562,6 @@ void appendToOptions(Options options) {
   private String tag;
   private String etag;
   private Boolean validateOnly;
-  private Boolean withOptimisticLock;
   private Boolean withExcludeTxnFromChangeStreams;
   private Boolean dataBoostEnabled;
   private DirectedReadOptions directedReadOptions;
@@ -540,6 +571,7 @@ void appendToOptions(Options options) {
   private Boolean lastStatement;
   private IsolationLevel isolationLevel;
   private XGoogSpannerRequestId reqId;
+  private ReadLockMode readLockMode;
 
   // Construction is via factory methods below.
   private Options() {}
@@ -644,10 +676,6 @@ Boolean validateOnly() {
     return validateOnly;
   }
 
-  Boolean withOptimisticLock() {
-    return withOptimisticLock;
-  }
-
   Boolean withExcludeTxnFromChangeStreams() {
     return withExcludeTxnFromChangeStreams;
   }
@@ -704,6 +732,10 @@ IsolationLevel isolationLevel() {
     return isolationLevel;
   }
 
+  ReadLockMode readLockMode() {
+    return readLockMode;
+  }
+
   @Override
   public String toString() {
     StringBuilder b = new StringBuilder();
@@ -740,9 +772,6 @@ public String toString() {
     if (validateOnly != null) {
       b.append("validateOnly: ").append(validateOnly).append(' ');
     }
-    if (withOptimisticLock != null) {
-      b.append("withOptimisticLock: ").append(withOptimisticLock).append(' ');
-    }
     if (withExcludeTxnFromChangeStreams != null) {
       b.append("withExcludeTxnFromChangeStreams: ")
           .append(withExcludeTxnFromChangeStreams)
@@ -772,6 +801,9 @@ public String toString() {
     if (reqId != null) {
       b.append("requestId: ").append(reqId.toString());
     }
+    if (readLockMode != null) {
+      b.append("readLockMode: ").append(readLockMode).append(' ');
+    }
     return b.toString();
   }
 
@@ -807,15 +839,15 @@ public boolean equals(Object o) {
         && Objects.equals(tag(), that.tag())
         && Objects.equals(etag(), that.etag())
         && Objects.equals(validateOnly(), that.validateOnly())
-        && Objects.equals(withOptimisticLock(), that.withOptimisticLock())
         && Objects.equals(withExcludeTxnFromChangeStreams(), that.withExcludeTxnFromChangeStreams())
         && Objects.equals(dataBoostEnabled(), that.dataBoostEnabled())
         && Objects.equals(directedReadOptions(), that.directedReadOptions())
         && Objects.equals(orderBy(), that.orderBy())
         && Objects.equals(isLastStatement(), that.isLastStatement())
         && Objects.equals(lockHint(), that.lockHint())
         && Objects.equals(isolationLevel(), that.isolationLevel())
-        && Objects.equals(reqId(), that.reqId());
+        && Objects.equals(reqId(), that.reqId())
+        && Objects.equals(readLockMode(), that.readLockMode());
   }
 
   @Override
@@ -857,9 +889,6 @@ public int hashCode() {
     if (validateOnly != null) {
       result = 31 * result + validateOnly.hashCode();
     }
-    if (withOptimisticLock != null) {
-      result = 31 * result + withOptimisticLock.hashCode();
-    }
     if (withExcludeTxnFromChangeStreams != null) {
       result = 31 * result + withExcludeTxnFromChangeStreams.hashCode();
     }
@@ -887,6 +916,9 @@ public int hashCode() {
     if (reqId != null) {
       result = 31 * result + reqId.hashCode();
     }
+    if (readLockMode != null) {
+      result = 31 * result + readLockMode.hashCode();
+    }
     return result;
   }
 

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

@@ -45,6 +45,7 @@
 import com.google.spanner.v1.RequestOptions;
 import com.google.spanner.v1.Transaction;
 import com.google.spanner.v1.TransactionOptions;
+import com.google.spanner.v1.TransactionOptions.ReadWrite;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -76,16 +77,16 @@ static TransactionOptions createReadWriteTransactionOptions(
       transactionOptions.setExcludeTxnFromChangeStreams(true);
     }
     TransactionOptions.ReadWrite.Builder readWrite = TransactionOptions.ReadWrite.newBuilder();
-    if (options.withOptimisticLock() == Boolean.TRUE) {
-      readWrite.setReadLockMode(TransactionOptions.ReadWrite.ReadLockMode.OPTIMISTIC);
-    }
     if (previousTransactionId != null
         && previousTransactionId != com.google.protobuf.ByteString.EMPTY) {
       readWrite.setMultiplexedSessionPreviousTransactionId(previousTransactionId);
     }
     if (options.isolationLevel() != null) {
       transactionOptions.setIsolationLevel(options.isolationLevel());
     }
+    if (options.readLockMode() != null) {
+      readWrite.setReadLockMode(options.readLockMode());
+    }
     transactionOptions.setReadWrite(readWrite);
     return transactionOptions.build();
   }
@@ -283,6 +284,10 @@ public CommitResponse writeAtLeastOnceWithOptions(
     if (options.isolationLevel() != null) {
       transactionOptionsBuilder.setIsolationLevel(options.isolationLevel());
     }
+    if (options.readLockMode() != null) {
+      transactionOptionsBuilder.setReadWrite(
+          ReadWrite.newBuilder().setReadLockMode(options.readLockMode()));
+    }
     requestBuilder.setSingleUseTransaction(
         defaultTransactionOptions().toBuilder().mergeFrom(transactionOptionsBuilder.build()));
 

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

@@ -69,6 +69,8 @@
 import com.google.spanner.v1.SpannerGrpc;
 import com.google.spanner.v1.TransactionOptions;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.grpc.CallCredentials;
 import io.grpc.CompressorRegistry;
 import io.grpc.Context;
@@ -1700,6 +1702,7 @@ public Builder setEnableEndToEndTracing(boolean enableEndToEndTracing) {
      * <pre>{@code
      * DefaultReadWriteTransactionOptions options = DefaultReadWriteTransactionOptions.newBuilder()
      * .setIsolationLevel(IsolationLevel.SERIALIZABLE)
+     * .setReadLockMode(ReadLockMode.OPTIMISTIC)
      * .build();
      * }</pre>
      */
@@ -1724,6 +1727,13 @@ public DefaultReadWriteTransactionOptionsBuilder setIsolationLevel(
           return this;
         }
 
+        public DefaultReadWriteTransactionOptionsBuilder setReadLockMode(
+            ReadLockMode readLockMode) {
+          transactionOptionsBuilder.setReadWrite(
+              ReadWrite.newBuilder().setReadLockMode(readLockMode));
+          return this;
+        }
+
         public DefaultReadWriteTransactionOptions build() {
           return new DefaultReadWriteTransactionOptions(transactionOptionsBuilder.build());
         }