feat: add default_isolation_level connection property

Add a `default_isolation_level` property for the Connection API.
This property will be used by the JDBC driver and PGAdapter to
set a default isolation level for all read/write transactions
that are executed by a connection.

Support for setting an isolation level for a single transaction
will be added in a follow-up pull request.

Author: olavloite

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

@@ -33,6 +33,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.spanner.v1.DirectedReadOptions;
+import com.google.spanner.v1.TransactionOptions;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Duration;
@@ -381,6 +382,30 @@ public DirectedReadOptions convert(String value) {
       return null;
     }
   }
+  
+  /** Converter for converting strings to {@link com.google.spanner.v1.TransactionOptions.IsolationLevel} values. */
+  static class IsolationLevelConverter
+      implements ClientSideStatementValueConverter<TransactionOptions.IsolationLevel> {
+    static final IsolationLevelConverter INSTANCE = new IsolationLevelConverter();
+
+    private final CaseInsensitiveEnumMap<TransactionOptions.IsolationLevel> values =
+        new CaseInsensitiveEnumMap<>(TransactionOptions.IsolationLevel.class);
+
+    private IsolationLevelConverter() {}
+
+    /** Constructor needed for reflection. */
+    public IsolationLevelConverter(String allowedValues) {}
+
+    @Override
+    public Class<TransactionOptions.IsolationLevel> getParameterClass() {
+      return TransactionOptions.IsolationLevel.class;
+    }
+
+    @Override
+    public TransactionOptions.IsolationLevel convert(String value) {
+      return values.get(value);
+    }
+  }
 
   /** Converter for converting strings to {@link AutocommitDmlMode} values. */
   static class AutocommitDmlModeConverter

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

@@ -42,6 +42,7 @@
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ResultSetStats;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import java.time.Duration;
 import java.util.Iterator;
 import java.util.Set;
@@ -218,6 +219,12 @@ public interface Connection extends AutoCloseable {
 
   /** @return <code>true</code> if this connection is in read-only mode */
   boolean isReadOnly();
+  
+  /** Sets the default isolation level for read/write transactions for this connection. */
+  void setDefaultIsolationLevel(IsolationLevel isolationLevel);
+
+  /** Returns the default isolation level for read/write transactions for this connection. */
+  IsolationLevel getDefaultIsolationLevel();
 
   /**
    * Sets the duration the connection should wait before automatically aborting the execution of a

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

@@ -27,6 +27,7 @@
 import static com.google.cloud.spanner.connection.ConnectionProperties.AUTO_PARTITION_MODE;
 import static com.google.cloud.spanner.connection.ConnectionProperties.DATA_BOOST_ENABLED;
 import static com.google.cloud.spanner.connection.ConnectionProperties.DDL_IN_TRANSACTION_MODE;
+import static com.google.cloud.spanner.connection.ConnectionProperties.DEFAULT_ISOLATION_LEVEL;
 import static com.google.cloud.spanner.connection.ConnectionProperties.DEFAULT_SEQUENCE_KIND;
 import static com.google.cloud.spanner.connection.ConnectionProperties.DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE;
 import static com.google.cloud.spanner.connection.ConnectionProperties.DIRECTED_READ;
@@ -90,6 +91,7 @@
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.ExecuteSqlRequest.QueryOptions;
 import com.google.spanner.v1.ResultSetStats;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
@@ -478,6 +480,7 @@ private void reset(Context context, boolean inTransaction) {
     this.connectionState.resetValue(RETRY_ABORTS_INTERNALLY, context, 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_ONLY_STALENESS, context, inTransaction);
     this.connectionState.resetValue(OPTIMIZER_VERSION, context, inTransaction);
     this.connectionState.resetValue(OPTIMIZER_STATISTICS_PACKAGE, context, inTransaction);
@@ -635,6 +638,22 @@ public boolean isReadOnly() {
     return getConnectionPropertyValue(READONLY);
   }
 
+  @Override
+  public void setDefaultIsolationLevel(IsolationLevel isolationLevel) {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    ConnectionPreconditions.checkState(!isBatchActive(), "Cannot default isolation level while in a batch");
+    ConnectionPreconditions.checkState(
+        !isTransactionStarted(), "Cannot set default isolation level while a transaction is active");
+    setConnectionPropertyValue(DEFAULT_ISOLATION_LEVEL, isolationLevel);
+    clearLastTransactionAndSetDefaultTransactionOptions();
+  }
+
+  @Override
+  public IsolationLevel getDefaultIsolationLevel() {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    return getConnectionPropertyValue(DEFAULT_ISOLATION_LEVEL);
+  }
+
   private void clearLastTransactionAndSetDefaultTransactionOptions() {
     setDefaultTransactionOptions();
     this.currentUnitOfWork = null;
@@ -2196,6 +2215,7 @@ UnitOfWork createNewUnitOfWork(
               .setUsesEmulator(options.usesEmulator())
               .setUseAutoSavepointsForEmulator(options.useAutoSavepointsForEmulator())
               .setDatabaseClient(dbClient)
+              .setIsolationLevel(getConnectionPropertyValue(DEFAULT_ISOLATION_LEVEL))
               .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

@@ -112,6 +112,7 @@
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.DdlInTransactionModeConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.DialectConverter;
 import com.google.cloud.spanner.connection.ClientSideStatementValueConverters.DurationConverter;
+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.ReadOnlyStalenessConverter;
@@ -123,7 +124,9 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.spanner.v1.DirectedReadOptions;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import java.time.Duration;
+import java.util.Arrays;
 
 /**
  * Utility class that defines all known connection properties. This class will eventually replace
@@ -401,13 +404,24 @@ public class ConnectionProperties {
           BOOLEANS,
           BooleanConverter.INSTANCE,
           Context.USER);
+  static final ConnectionProperty<IsolationLevel> DEFAULT_ISOLATION_LEVEL =
+      create(
+          "default_isolation_level",
+          "The transaction isolation level that is used by default for read/write transactions. " 
+              + "The default is isolation_level_unspecified, which means that the connection will use the " 
+              + "default isolation level of the database that it is connected to.",
+          IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
+          new IsolationLevel[]{IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED, IsolationLevel.SERIALIZABLE, IsolationLevel.REPEATABLE_READ},
+          IsolationLevelConverter.INSTANCE,
+          Context.USER);
   static final ConnectionProperty<AutocommitDmlMode> AUTOCOMMIT_DML_MODE =
       create(
           "autocommit_dml_mode",
           "Determines the transaction type that is used to execute "
               + "DML statements when the connection is in auto-commit mode.",
           AutocommitDmlMode.TRANSACTIONAL,
-          AutocommitDmlMode.values(),
+          // Add 'null' as a valid value.
+          Arrays.copyOf(AutocommitDmlMode.values(), AutocommitDmlMode.values().length + 1),
           AutocommitDmlModeConverter.INSTANCE,
           Context.USER);
   static final ConnectionProperty<Boolean> RETRY_ABORTS_INTERNALLY =
@@ -523,7 +537,8 @@ public class ConnectionProperties {
           RPC_PRIORITY_NAME,
           "Sets the priority for all RPC invocations from this connection (HIGH/MEDIUM/LOW). The default is HIGH.",
           DEFAULT_RPC_PRIORITY,
-          RpcPriority.values(),
+          // Add 'null' as a valid value.
+          Arrays.copyOf(RpcPriority.values(), RpcPriority.values().length + 1),
           RpcPriorityConverter.INSTANCE,
           Context.USER);
   static final ConnectionProperty<SavepointSupport> SAVEPOINT_SUPPORT =

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

@@ -27,10 +27,12 @@
 import com.google.cloud.spanner.connection.ConnectionProperty.Context;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Suppliers;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
@@ -233,6 +235,7 @@ private <T> void internalSetValue(
       T value,
       Map<String, ConnectionPropertyValue<?>> currentProperties,
       Context context) {
+    checkValidValue(property, value);
     ConnectionPropertyValue<T> newValue = cast(currentProperties.get(property.getKey()));
     if (newValue == null) {
       ConnectionPropertyValue<T> existingValue = cast(properties.get(property.getKey()));
@@ -248,6 +251,22 @@ private <T> void internalSetValue(
     newValue.setValue(value, context);
     currentProperties.put(property.getKey(), newValue);
   }
+  
+  static <T> void checkValidValue(ConnectionProperty<T> property, T value) {
+    if (property.getValidValues() == null || property.getValidValues().length == 0) {
+      return;
+    }
+    if (Arrays.stream(property.getValidValues()).noneMatch(validValue -> Objects.equals(validValue, value))) {
+      throw invalidParamValueError(property, value);
+    }
+  }
+  
+  /** Creates an exception for an invalid value for a connection property. */
+  static <T> SpannerException invalidParamValueError(ConnectionProperty<T> property, T value) {
+    return SpannerExceptionFactory.newSpannerException(
+        ErrorCode.INVALID_ARGUMENT,
+        String.format("invalid value \"%s\" for configuration property \"%s\"", value, property));
+  }
 
   /** Creates an exception for an unknown connection property. */
   static SpannerException unknownParamError(ConnectionProperty<?> property) {

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

@@ -60,6 +60,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.v1.SpannerGrpc;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.context.Scope;
 import java.time.Duration;
@@ -151,6 +152,7 @@ class ReadWriteTransaction extends AbstractMultiUseTransaction {
   private final long keepAliveIntervalMillis;
   private final ReentrantLock keepAliveLock;
   private final SavepointSupport savepointSupport;
+  @Nonnull private final IsolationLevel isolationLevel;
   private int transactionRetryAttempts;
   private int successfulRetries;
   private volatile ApiFuture<TransactionContext> txContextFuture;
@@ -202,6 +204,7 @@ static class Builder extends AbstractMultiUseTransaction.Builder<Builder, ReadWr
     private boolean returnCommitStats;
     private Duration maxCommitDelay;
     private SavepointSupport savepointSupport;
+    private IsolationLevel isolationLevel;
 
     private Builder() {}
 
@@ -250,6 +253,11 @@ Builder setSavepointSupport(SavepointSupport savepointSupport) {
       this.savepointSupport = savepointSupport;
       return this;
     }
+    
+    Builder setIsolationLevel(IsolationLevel isolationLevel) {
+      this.isolationLevel = Preconditions.checkNotNull(isolationLevel);
+      return this;
+    }
 
     @Override
     ReadWriteTransaction build() {
@@ -259,6 +267,7 @@ ReadWriteTransaction build() {
       Preconditions.checkState(
           hasTransactionRetryListeners(), "TransactionRetryListeners are not specified");
       Preconditions.checkState(savepointSupport != null, "SavepointSupport is not specified");
+      Preconditions.checkState(isolationLevel != null, "IsolationLevel is not specified");
       return new ReadWriteTransaction(this);
     }
   }
@@ -293,6 +302,7 @@ private ReadWriteTransaction(Builder builder) {
     this.keepAliveLock = this.keepTransactionAlive ? new ReentrantLock() : null;
     this.retryAbortsInternally = builder.retryAbortsInternally;
     this.savepointSupport = builder.savepointSupport;
+    this.isolationLevel = Preconditions.checkNotNull(builder.isolationLevel);
     this.transactionOptions = extractOptions(builder);
   }
 
@@ -313,6 +323,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (this.rpcPriority != null) {
       numOptions++;
     }
+    if (this.isolationLevel != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
+      numOptions++;
+    }
     TransactionOption[] options = new TransactionOption[numOptions];
     int index = 0;
     if (builder.returnCommitStats) {
@@ -330,6 +343,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (this.rpcPriority != null) {
       options[index++] = Options.priority(this.rpcPriority);
     }
+    if (this.isolationLevel != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
+      options[index++] = Options.isolationLevel(this.isolationLevel);
+    }
     return options;
   }
 

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

@@ -19,6 +19,7 @@
 import static com.google.cloud.spanner.connection.AbstractStatementParser.COMMIT_STATEMENT;
 import static com.google.cloud.spanner.connection.AbstractStatementParser.RUN_BATCH_STATEMENT;
 import static com.google.cloud.spanner.connection.ConnectionProperties.AUTOCOMMIT_DML_MODE;
+import static com.google.cloud.spanner.connection.ConnectionProperties.DEFAULT_ISOLATION_LEVEL;
 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;
@@ -60,6 +61,7 @@
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.admin.database.v1.DatabaseAdminGrpc;
 import com.google.spanner.v1.SpannerGrpc;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
 import io.opentelemetry.context.Scope;
 import java.util.Arrays;
 import java.util.UUID;
@@ -508,6 +510,9 @@ private TransactionRunner createWriteTransaction() {
     if (connectionState.getValue(MAX_COMMIT_DELAY).getValue() != null) {
       numOptions++;
     }
+    if (connectionState.getValue(DEFAULT_ISOLATION_LEVEL).getValue() != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
+      numOptions++;
+    }
     if (numOptions == 0) {
       return dbClient.readWriteTransaction();
     }
@@ -526,6 +531,9 @@ private TransactionRunner createWriteTransaction() {
       options[index++] =
           Options.maxCommitDelay(connectionState.getValue(MAX_COMMIT_DELAY).getValue());
     }
+    if (connectionState.getValue(DEFAULT_ISOLATION_LEVEL).getValue() != IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
+      options[index++] = Options.isolationLevel(connectionState.getValue(DEFAULT_ISOLATION_LEVEL).getValue());
+    }
     return dbClient.readWriteTransaction(options);
   }