feat: read_lock_mode support for connections

Add read_lock_mode support for the Connection API.

Author: olavloite

google-cloud-spanner/clirr-ignored-differences.xml

@@ -1033,4 +1033,14 @@
     <className>com/google/cloud/spanner/SpannerOptions$SpannerEnvironment</className>
     <method>boolean isEnableDirectAccess()</method>
   </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>void setReadLockMode(com.google.spanner.v1.TransactionOptions$ReadWrite$ReadLockMode)</method>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>com.google.spanner.v1.TransactionOptions$ReadWrite$ReadLockMode getReadLockMode()</method>
+  </difference>
 </differences>

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ClientSideStatementValueConverters.java

@@ -34,6 +34,7 @@
 import com.google.common.base.Strings;
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.TransactionOptions;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Duration;
@@ -425,6 +426,36 @@ public TransactionOptions.IsolationLevel convert(String value) {
     }
   }
 
+  /**
+   * Converter for converting strings to {@link
+   * com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode} values.
+   */
+  static class ReadLockModeConverter implements ClientSideStatementValueConverter<ReadLockMode> {
+    static final ReadLockModeConverter INSTANCE = new ReadLockModeConverter();
+
+    private final CaseInsensitiveEnumMap<ReadLockMode> values =
+        new CaseInsensitiveEnumMap<>(ReadLockMode.class);
+
+    ReadLockModeConverter() {}
+
+    /** Constructor needed for reflection. */
+    public ReadLockModeConverter(String allowedValues) {}
+
+    @Override
+    public Class<ReadLockMode> getParameterClass() {
+      return ReadLockMode.class;
+    }
+
+    @Override
+    public ReadLockMode convert(String value) {
+      if (value != null && value.equalsIgnoreCase("unspecified")) {
+        // Allow 'unspecified' to be used in addition to 'read_lock_mode_unspecified'.
+        value = ReadLockMode.READ_LOCK_MODE_UNSPECIFIED.name();
+      }
+      return values.get(value);
+    }
+  }
+
   /** Converter for converting strings to {@link AutocommitDmlMode} values. */
   static class AutocommitDmlModeConverter
       implements ClientSideStatementValueConverter<AutocommitDmlMode> {

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/Connection.java

@@ -43,6 +43,7 @@
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ResultSetStats;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.time.Duration;
 import java.util.Iterator;
 import java.util.Set;
@@ -232,6 +233,12 @@ public interface Connection extends AutoCloseable {
   /** Returns the default isolation level for read/write transactions for this connection. */
   IsolationLevel getDefaultIsolationLevel();
 
+  /** Sets the read lock mode for read/write transactions for this connection. */
+  void setReadLockMode(ReadLockMode readLockMode);
+
+  /** Returns the read lock mode for read/write transactions for this connection. */
+  ReadLockMode getReadLockMode();
+
   /**
    * Sets the duration the connection should wait before automatically aborting the execution of a
    * statement. The default is no timeout. Statement timeouts are applied all types of statements,

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java

@@ -38,6 +38,7 @@
 import static com.google.cloud.spanner.connection.ConnectionProperties.OPTIMIZER_STATISTICS_PACKAGE;
 import static com.google.cloud.spanner.connection.ConnectionProperties.OPTIMIZER_VERSION;
 import static com.google.cloud.spanner.connection.ConnectionProperties.READONLY;
+import static com.google.cloud.spanner.connection.ConnectionProperties.READ_LOCK_MODE;
 import static com.google.cloud.spanner.connection.ConnectionProperties.READ_ONLY_STALENESS;
 import static com.google.cloud.spanner.connection.ConnectionProperties.RETRY_ABORTS_INTERNALLY;
 import static com.google.cloud.spanner.connection.ConnectionProperties.RETURN_COMMIT_STATS;
@@ -92,6 +93,7 @@
 import com.google.spanner.v1.ExecuteSqlRequest.QueryOptions;
 import com.google.spanner.v1.ResultSetStats;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
@@ -486,6 +488,7 @@ private void reset(Context context, boolean inTransaction) {
     this.connectionState.resetValue(AUTOCOMMIT, context, inTransaction);
     this.connectionState.resetValue(READONLY, context, inTransaction);
     this.connectionState.resetValue(DEFAULT_ISOLATION_LEVEL, context, inTransaction);
+    this.connectionState.resetValue(READ_LOCK_MODE, context, inTransaction);
     this.connectionState.resetValue(READ_ONLY_STALENESS, context, inTransaction);
     this.connectionState.resetValue(OPTIMIZER_VERSION, context, inTransaction);
     this.connectionState.resetValue(OPTIMIZER_STATISTICS_PACKAGE, context, inTransaction);
@@ -668,6 +671,18 @@ private void clearLastTransactionAndSetDefaultTransactionOptions(IsolationLevel
     this.currentUnitOfWork = null;
   }
 
+  @Override
+  public void setReadLockMode(ReadLockMode readLockMode) {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    setConnectionPropertyValue(READ_LOCK_MODE, readLockMode);
+  }
+
+  @Override
+  public ReadLockMode getReadLockMode() {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    return getConnectionPropertyValue(READ_LOCK_MODE);
+  }
+
   @Override
   public void setAutocommitDmlMode(AutocommitDmlMode mode) {
     Preconditions.checkNotNull(mode);
@@ -2255,6 +2270,7 @@ UnitOfWork createNewUnitOfWork(
               .setUseAutoSavepointsForEmulator(options.useAutoSavepointsForEmulator())
               .setDatabaseClient(dbClient)
               .setIsolationLevel(transactionIsolationLevel)
+              .setReadLockMode(getConnectionPropertyValue(READ_LOCK_MODE))
               .setDelayTransactionStartUntilFirstWrite(
                   getConnectionPropertyValue(DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE))
               .setKeepTransactionAlive(getConnectionPropertyValue(KEEP_TRANSACTION_ALIVE))

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionProperties.java

@@ -115,6 +115,7 @@
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.IsolationLevelConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.LongConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.NonNegativeIntegerConverter;
+import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.ReadLockModeConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.ReadOnlyStalenessConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.RpcPriorityConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.SavepointSupportConverter;
@@ -125,8 +126,10 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.stream.Collectors;
 
 /** Utility class that defines all known connection properties. */
 public class ConnectionProperties {
@@ -451,6 +454,30 @@ public class ConnectionProperties {
           },
           IsolationLevelConverter.INSTANCE,
           Context.USER);
+  static final ConnectionProperty<ReadLockMode> READ_LOCK_MODE =
+      create(
+          "read_lock_mode",
+          "This option controls the locking behavior for read operations and queries within a read/write transaction. "
+              + "It works in conjunction with the transaction's isolation level.\n\n"
+              + "PESSIMISTIC: Read locks are acquired immediately on read. This mode only applies to SERIALIZABLE isolation. "
+              + "This mode prevents concurrent modifications by locking data throughout the transaction. This reduces commit-time "
+              + "aborts due to conflicts, but can increase how long transactions wait for locks and the overall contention.\n\n"
+              + "OPTIMISTIC: Locks for reads within the transaction are not acquired on read. Instead, the locks are acquired on "
+              + "commit to validate that read/queried data has not changed since the transaction started. If a conflict is "
+              + "detected, the transaction will fail. This mode only applies to SERIALIZABLE isolation. This mode defers locking "
+              + "until commit, which can reduce contention and improve throughput. However, be aware that this increases the "
+              + "risk of transaction aborts if there's significant write competition on the same data.\n\n"
+              + "READ_LOCK_MODE_UNSPECIFIED: This is the default if no mode is set. The locking behavior depends on the isolation level:\n\n"
+              + "REPEATABLE_READ: Locking semantics default to OPTIMISTIC. However, validation checks at commit are only "
+              + "performed for queries using SELECT FOR UPDATE, statements with {@code LOCK_SCANNED_RANGES} hints, and DML statements.\n\n"
+              + "For all other isolation levels: If the read lock mode is not set, it defaults to PESSIMISTIC locking.",
+          ReadLockMode.READ_LOCK_MODE_UNSPECIFIED,
+          Arrays.stream(ReadLockMode.values())
+              .filter(mode -> !mode.equals(ReadLockMode.UNRECOGNIZED))
+              .collect(Collectors.toList())
+              .toArray(new ReadLockMode[0]),
+          ReadLockModeConverter.INSTANCE,
+          Context.USER);
   static final ConnectionProperty<AutocommitDmlMode> AUTOCOMMIT_DML_MODE =
       create(
           "autocommit_dml_mode",

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionStatementExecutor.java

@@ -23,6 +23,7 @@
 import com.google.cloud.spanner.connection.PgTransactionMode.IsolationLevel;
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.TransactionOptions;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.time.Duration;
 
 /**
@@ -190,4 +191,8 @@ StatementResult statementSetPgSessionCharacteristicsTransactionMode(
   StatementResult statementSetAutoBatchDmlUpdateCountVerification(Boolean verification);
 
   StatementResult statementShowAutoBatchDmlUpdateCountVerification();
+
+  StatementResult statementSetReadLockMode(ReadLockMode readLockMode);
+
+  StatementResult statementShowReadLockMode();
 }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionStatementExecutorImpl.java

@@ -43,6 +43,7 @@
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_PROTO_DESCRIPTORS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_PROTO_DESCRIPTORS_FILE_PATH;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_READONLY;
+import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_READ_LOCK_MODE;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_READ_ONLY_STALENESS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_RETRY_ABORTS_INTERNALLY;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_RETURN_COMMIT_STATS;
@@ -73,6 +74,7 @@
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_PROTO_DESCRIPTORS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_PROTO_DESCRIPTORS_FILE_PATH;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_READONLY;
+import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_READ_LOCK_MODE;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_READ_ONLY_STALENESS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_READ_TIMESTAMP;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_RETRY_ABORTS_INTERNALLY;
@@ -113,6 +115,7 @@
 import com.google.spanner.v1.QueryPlan;
 import com.google.spanner.v1.RequestOptions;
 import com.google.spanner.v1.TransactionOptions;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -607,6 +610,20 @@ public StatementResult statementShowSavepointSupport() {
         SHOW_SAVEPOINT_SUPPORT);
   }
 
+  @Override
+  public StatementResult statementSetReadLockMode(ReadLockMode readLockMode) {
+    getConnection().setReadLockMode(readLockMode);
+    return noResult(SET_READ_LOCK_MODE);
+  }
+
+  @Override
+  public StatementResult statementShowReadLockMode() {
+    return resultSet(
+        String.format("%sREAD_LOCK_MODE", getNamespace(connection.getDialect())),
+        getConnection().getReadLockMode(),
+        SHOW_READ_LOCK_MODE);
+  }
+
   @Override
   public StatementResult statementShowTransactionIsolationLevel() {
     return resultSet("transaction_isolation", "serializable", SHOW_TRANSACTION_ISOLATION_LEVEL);

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java

@@ -61,6 +61,7 @@
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.v1.SpannerGrpc;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.context.Scope;
 import java.time.Duration;
@@ -155,6 +156,7 @@ class ReadWriteTransaction extends AbstractMultiUseTransaction {
   private final ReentrantLock keepAliveLock;
   private final SavepointSupport savepointSupport;
   @Nonnull private final IsolationLevel isolationLevel;
+  private final ReadLockMode readLockMode;
   private int transactionRetryAttempts;
   private int successfulRetries;
   private volatile ApiFuture<TransactionContext> txContextFuture;
@@ -207,6 +209,7 @@ static class Builder extends AbstractMultiUseTransaction.Builder<Builder, ReadWr
     private Duration maxCommitDelay;
     private SavepointSupport savepointSupport;
     private IsolationLevel isolationLevel;
+    private ReadLockMode readLockMode = ReadLockMode.READ_LOCK_MODE_UNSPECIFIED;
 
     private Builder() {}
 
@@ -261,6 +264,11 @@ Builder setIsolationLevel(IsolationLevel isolationLevel) {
       return this;
     }
 
+    Builder setReadLockMode(ReadLockMode readLockMode) {
+      this.readLockMode = Preconditions.checkNotNull(readLockMode);
+      return this;
+    }
+
     @Override
     ReadWriteTransaction build() {
       Preconditions.checkState(dbClient != null, "No DatabaseClient client specified");
@@ -305,6 +313,7 @@ private ReadWriteTransaction(Builder builder) {
     this.retryAbortsInternally = builder.retryAbortsInternally;
     this.savepointSupport = builder.savepointSupport;
     this.isolationLevel = Preconditions.checkNotNull(builder.isolationLevel);
+    this.readLockMode = Preconditions.checkNotNull(builder.readLockMode);
     this.transactionOptions = extractOptions(builder);
   }
 
@@ -328,6 +337,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (this.isolationLevel != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
       numOptions++;
     }
+    if (this.readLockMode != ReadLockMode.READ_LOCK_MODE_UNSPECIFIED) {
+      numOptions++;
+    }
     TransactionOption[] options = new TransactionOption[numOptions];
     int index = 0;
     if (builder.returnCommitStats) {
@@ -348,6 +360,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (this.isolationLevel != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
       options[index++] = Options.isolationLevel(this.isolationLevel);
     }
+    if (this.readLockMode != ReadLockMode.READ_LOCK_MODE_UNSPECIFIED) {
+      options[index++] = Options.readLockMode(this.readLockMode);
+    }
     return options;
   }
 

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/SingleUseTransaction.java

@@ -23,6 +23,7 @@
 import static com.google.cloud.spanner.connection.ConnectionProperties.DEFAULT_SEQUENCE_KIND;
 import static com.google.cloud.spanner.connection.ConnectionProperties.MAX_COMMIT_DELAY;
 import static com.google.cloud.spanner.connection.ConnectionProperties.READONLY;
+import static com.google.cloud.spanner.connection.ConnectionProperties.READ_LOCK_MODE;
 import static com.google.cloud.spanner.connection.ConnectionProperties.READ_ONLY_STALENESS;
 import static com.google.cloud.spanner.connection.ConnectionProperties.RETURN_COMMIT_STATS;
 import static com.google.cloud.spanner.connection.DdlClient.isCreateDatabaseStatement;
@@ -62,6 +63,7 @@
 import com.google.spanner.admin.database.v1.DatabaseAdminGrpc;
 import com.google.spanner.v1.SpannerGrpc;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.opentelemetry.context.Scope;
 import java.util.Arrays;
 import java.util.UUID;
@@ -514,6 +516,10 @@ private TransactionRunner createWriteTransaction() {
         != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
       numOptions++;
     }
+    if (connectionState.getValue(READ_LOCK_MODE).getValue()
+        != ReadLockMode.READ_LOCK_MODE_UNSPECIFIED) {
+      numOptions++;
+    }
     if (numOptions == 0) {
       return dbClient.readWriteTransaction();
     }
@@ -537,6 +543,10 @@ private TransactionRunner createWriteTransaction() {
       options[index++] =
           Options.isolationLevel(connectionState.getValue(DEFAULT_ISOLATION_LEVEL).getValue());
     }
+    if (connectionState.getValue(READ_LOCK_MODE).getValue()
+        != ReadLockMode.READ_LOCK_MODE_UNSPECIFIED) {
+      options[index++] = Options.readLockMode(connectionState.getValue(READ_LOCK_MODE).getValue());
+    }
     return dbClient.readWriteTransaction(options);
   }
 

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/StatementResult.java

@@ -120,6 +120,8 @@ enum ClientSideStatementType {
     SHOW_AUTO_BATCH_DML_UPDATE_COUNT,
     SET_AUTO_BATCH_DML_UPDATE_COUNT_VERIFICATION,
     SHOW_AUTO_BATCH_DML_UPDATE_COUNT_VERIFICATION,
+    SHOW_READ_LOCK_MODE,
+    SET_READ_LOCK_MODE,
   }
 
   /**