feat(spanner): implement transaction routing logic based on database metadata isolation levels and lock modes

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner;
+
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
+import java.util.Objects;
+
+/**
+ * Internal container for dynamic database-level defaults queried from
+ * INFORMATION_SCHEMA.DATABASE_OPTIONS. Holds the database dialect, default transaction isolation
+ * level, and default read lock mode.
+ */
+final class DatabaseMetadata {
+  private final Dialect dialect;
+  private final IsolationLevel isolationLevel;
+  private final ReadLockMode readLockMode;
+
+  DatabaseMetadata(Dialect dialect, IsolationLevel isolationLevel, ReadLockMode readLockMode) {
+    this.dialect = Objects.requireNonNull(dialect);
+    this.isolationLevel = Objects.requireNonNull(isolationLevel);
+    this.readLockMode = Objects.requireNonNull(readLockMode);
+  }
+
+  Dialect getDialect() {
+    return dialect;
+  }
+
+  IsolationLevel getIsolationLevel() {
+    return isolationLevel;
+  }
+
+  ReadLockMode getReadLockMode() {
+    return readLockMode;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    DatabaseMetadata that = (DatabaseMetadata) o;
+    return dialect == that.dialect
+        && isolationLevel == that.isolationLevel
+        && readLockMode == that.readLockMode;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(dialect, isolationLevel, readLockMode);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "DatabaseMetadata{dialect=%s, isolation=%s, lockMode=%s}",
+        dialect, isolationLevel, readLockMode);
+  }
+}

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

@@ -31,6 +31,8 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.spanner.v1.BatchWriteResponse;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
@@ -63,14 +65,30 @@ final class MultiplexedSessionDatabaseClient extends AbstractMultiplexedSessionD
    */
   private static final int MAX_INITIAL_CREATE_SESSION_ATTEMPTS = 10;
 
+  /**
+   * Statement used to query database-level default options from
+   * INFORMATION_SCHEMA.DATABASE_OPTIONS. This retrieves 'default_transaction_isolation',
+   * 'default_read_lock_mode', and 'database_dialect' so that the client can correctly configure
+   * transaction routing (e.g. Leader-Aware Routing) and isolation level behaviors without relying
+   * solely on client-side hardcoded defaults.
+   */
   @VisibleForTesting
-  static final Statement DETERMINE_DIALECT_STATEMENT =
+  static final Statement DETERMINE_METADATA_STATEMENT =
       Statement.newBuilder(
-              "select option_value "
-                  + "from information_schema.database_options "
-                  + "where option_name='database_dialect'")
+              "SELECT OPTION_NAME, OPTION_VALUE "
+                  + "FROM INFORMATION_SCHEMA.DATABASE_OPTIONS "
+                  + "WHERE OPTION_NAME IN ('default_transaction_isolation', "
+                  + "'default_read_lock_mode', 'database_dialect')")
           .build();
 
+  static final String OPTION_DATABASE_DIALECT = "database_dialect";
+  static final String OPTION_DEFAULT_TRANSACTION_ISOLATION = "default_transaction_isolation";
+  static final String OPTION_DEFAULT_READ_LOCK_MODE = "default_read_lock_mode";
+
+  static final String ISOLATION_LEVEL_REPEATABLE_READ = "repeatable read";
+  static final String READ_LOCK_MODE_OPTIMISTIC = "optimistic";
+  static final String READ_LOCK_MODE_PESSIMISTIC = "pessimistic";
+
   /**
    * Represents a single transaction on a multiplexed session. This can be both a single-use or
    * multi-use transaction, and both read/write or read-only transaction. This can be compared to a
@@ -274,13 +292,8 @@ public void onSessionReady(SessionImpl session) {
             // only start the maintainer if we actually managed to create a session in the first
             // place.
             maintainer.start();
-            if (sessionClient
-                .getSpanner()
-                .getOptions()
-                .getSessionPoolOptions()
-                .isAutoDetectDialect()) {
-              MAINTAINER_SERVICE.submit(() -> getDialect());
-            }
+            MAINTAINER_SERVICE.submit(
+                () -> session.getSessionReference().setDatabaseMetadata(getDatabaseMetadata()));
           }
 
           @Override
@@ -371,6 +384,12 @@ AtomicLong getNumSessionsReleased() {
     return this.numSessionsReleased;
   }
 
+  @VisibleForTesting
+  void resetAcquiredAndReleasedCounts() {
+    this.numSessionsAcquired.set(0L);
+    this.numSessionsReleased.set(0L);
+  }
+
   void close() {
     boolean releaseChannelUsage = false;
     synchronized (this) {
@@ -473,32 +492,71 @@ private int getSingleUseChannelHint() {
     }
   }
 
-  private final AbstractLazyInitializer<Dialect> dialectSupplier =
-      new AbstractLazyInitializer<Dialect>() {
+  private static IsolationLevel parseIsolationLevel(String value) {
+    return ISOLATION_LEVEL_REPEATABLE_READ.equalsIgnoreCase(value)
+        ? IsolationLevel.REPEATABLE_READ
+        : IsolationLevel.SERIALIZABLE;
+  }
+
+  private static ReadLockMode parseReadLockMode(String value) {
+    if (READ_LOCK_MODE_OPTIMISTIC.equalsIgnoreCase(value)) {
+      return ReadLockMode.OPTIMISTIC;
+    } else if (READ_LOCK_MODE_PESSIMISTIC.equalsIgnoreCase(value)) {
+      return ReadLockMode.PESSIMISTIC;
+    }
+    return ReadLockMode.READ_LOCK_MODE_UNSPECIFIED;
+  }
+
+  /**
+   * Lazily initializes and caches {@link DatabaseMetadata} (dialect, default isolation level, and
+   * read lock mode). Introspects the database options once and attaches the resolved metadata to
+   * the current multiplexed {@link SessionReference} so subsequent transactions can resolve their
+   * effective modes.
+   */
+  private final AbstractLazyInitializer<DatabaseMetadata> metadataSupplier =
+      new AbstractLazyInitializer<DatabaseMetadata>() {
         @Override
-        protected Dialect initialize() {
-          try (ResultSet dialectResultSet = singleUse().executeQuery(DETERMINE_DIALECT_STATEMENT)) {
-            if (dialectResultSet.next()) {
-              return Dialect.fromName(dialectResultSet.getString(0));
+        protected DatabaseMetadata initialize() {
+          Dialect dialect = Dialect.GOOGLE_STANDARD_SQL;
+          IsolationLevel isolationLevel = IsolationLevel.SERIALIZABLE;
+          ReadLockMode readLockMode = ReadLockMode.READ_LOCK_MODE_UNSPECIFIED;
+
+          numSessionsAcquired.decrementAndGet();
+          try (ResultSet resultSet = singleUse().executeQuery(DETERMINE_METADATA_STATEMENT)) {
+            while (resultSet.next()) {
+              String name = resultSet.getString(0);
+              String value = resultSet.getString(1);
+              if (OPTION_DATABASE_DIALECT.equalsIgnoreCase(name)) {
+                dialect = Dialect.fromName(value);
+              } else if (OPTION_DEFAULT_TRANSACTION_ISOLATION.equalsIgnoreCase(name)) {
+                isolationLevel = parseIsolationLevel(value);
+              } else if (OPTION_DEFAULT_READ_LOCK_MODE.equalsIgnoreCase(name)) {
+                readLockMode = parseReadLockMode(value);
+              }
             }
+          } finally {
+            numSessionsReleased.decrementAndGet();
           }
-          // This should not really happen, but it is the safest fallback value.
-          return Dialect.GOOGLE_STANDARD_SQL;
+          return new DatabaseMetadata(dialect, isolationLevel, readLockMode);
         }
       };
 
-  @Override
-  public Dialect getDialect() {
+  DatabaseMetadata getDatabaseMetadata() {
     try {
-      return dialectSupplier.get();
+      return metadataSupplier.get();
     } catch (Exception exception) {
       throw SpannerExceptionFactory.asSpannerException(exception);
     }
   }
 
+  @Override
+  public Dialect getDialect() {
+    return getDatabaseMetadata().getDialect();
+  }
+
   Future<Dialect> getDialectAsync() {
     try {
-      return MAINTAINER_SERVICE.submit(dialectSupplier::get);
+      return MAINTAINER_SERVICE.submit(() -> getDialect());
     } catch (Exception exception) {
       throw SpannerExceptionFactory.asSpannerException(exception);
     }
@@ -659,8 +717,10 @@ void maintain() {
             new SessionConsumer() {
               @Override
               public void onSessionReady(SessionImpl session) {
-                multiplexedSessionReference.set(
-                    ApiFutures.immediateFuture(session.getSessionReference()));
+                SessionReference sessionRef = session.getSessionReference();
+                multiplexedSessionReference.set(ApiFutures.immediateFuture(sessionRef));
+                MAINTAINER_SERVICE.submit(
+                    () -> sessionRef.setDatabaseMetadata(getDatabaseMetadata()));
                 expirationDate.set(
                     clock
                         .instant()

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

@@ -38,6 +38,7 @@ class SessionReference {
   private volatile Instant lastUseTime;
   @Nullable private final Instant createTime;
   private final boolean isMultiplexed;
+  private volatile DatabaseMetadata databaseMetadata;
 
   SessionReference(String name, @Nullable String databaseRole, Map<SpannerRpc.Option, ?> options) {
     this.options = options;
@@ -92,6 +93,15 @@ boolean getIsMultiplexed() {
     return isMultiplexed;
   }
 
+  @Nullable
+  DatabaseMetadata getDatabaseMetadata() {
+    return databaseMetadata;
+  }
+
+  void setDatabaseMetadata(DatabaseMetadata databaseMetadata) {
+    this.databaseMetadata = databaseMetadata;
+  }
+
   void markUsed(Instant instant) {
     lastUseTime = instant;
   }

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

@@ -54,6 +54,8 @@
 import com.google.spanner.v1.RollbackRequest;
 import com.google.spanner.v1.Transaction;
 import com.google.spanner.v1.TransactionOptions;
+import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import com.google.spanner.v1.TransactionSelector;
 import java.util.ArrayList;
 import java.util.List;
@@ -223,8 +225,19 @@ public void removeListener(Runnable listener) {
     private CommitResponse commitResponse;
     private final Clock clock;
 
+    private final boolean routeToLeader;
     private final Map<SpannerRpc.Option, ?> channelHint;
 
+    private static final class TransactionMode {
+      private final IsolationLevel isolationLevel;
+      private final ReadLockMode readLockMode;
+
+      TransactionMode(IsolationLevel isolationLevel, ReadLockMode readLockMode) {
+        this.isolationLevel = isolationLevel;
+        this.readLockMode = readLockMode;
+      }
+    }
+
     private TransactionContextImpl(Builder builder) {
       super(builder);
       this.transactionId = builder.transactionId;
@@ -238,6 +251,81 @@ private TransactionContextImpl(Builder builder) {
               ThreadLocalRandom.current().nextLong(Long.MAX_VALUE),
               session.getSpanner().getOptions().isGrpcGcpExtensionEnabled());
       this.previousTransactionId = builder.previousTransactionId;
+
+      TransactionMode effectiveMode = resolveEffectiveMode(this.options);
+      this.routeToLeader = !canEnableLRYW(effectiveMode);
+    }
+
+    /**
+     * Resolves the effective isolation level and read lock mode using the following precedence: 1.
+     * Call-site options explicitly passed to the transaction (`callSite`). 2. Client-side static
+     * default transaction options configured on `SpannerOptions`. 3. Database-level defaults
+     * queried from `INFORMATION_SCHEMA.DATABASE_OPTIONS` (`dbDefaults`). 4. Hardcoded spanner
+     * defaults (`SERIALIZABLE` isolation level).
+     */
+    private TransactionMode resolveEffectiveMode(Options callSite) {
+      IsolationLevel isolationLevel = callSite.isolationLevel();
+      ReadLockMode readLockMode = callSite.readLockMode();
+
+      TransactionOptions defaultTxOptions = null;
+      if (session.getSpanner() != null && session.getSpanner().getOptions() != null) {
+        defaultTxOptions = session.getSpanner().getOptions().getDefaultTransactionOptions();
+      }
+      if (defaultTxOptions != null) {
+        if (isUnspecified(isolationLevel)) {
+          isolationLevel = defaultTxOptions.getIsolationLevel();
+        }
+        if (isUnspecified(readLockMode) && defaultTxOptions.hasReadWrite()) {
+          readLockMode = defaultTxOptions.getReadWrite().getReadLockMode();
+        }
+      }
+
+      DatabaseMetadata dbDefaults = null;
+      if (session.getSessionReference() != null) {
+        dbDefaults = session.getSessionReference().getDatabaseMetadata();
+      }
+      if (dbDefaults != null) {
+        if (isUnspecified(isolationLevel)) {
+          isolationLevel = dbDefaults.getIsolationLevel();
+        }
+        if (isUnspecified(readLockMode)) {
+          readLockMode = dbDefaults.getReadLockMode();
+        }
+      }
+
+      if (isUnspecified(isolationLevel)) {
+        isolationLevel = IsolationLevel.SERIALIZABLE;
+      }
+      // For REPEATABLE_READ, keep lock mode unspecified/null when not explicitly set so that
+      // canEnableLRYW evaluates to true for Leader-Routed Read-Your-Writes.
+      if (isUnspecified(readLockMode) && isolationLevel != IsolationLevel.REPEATABLE_READ) {
+        readLockMode = ReadLockMode.PESSIMISTIC;
+      }
+
+      return new TransactionMode(isolationLevel, readLockMode);
+    }
+
+    private static boolean isUnspecified(IsolationLevel level) {
+      return level == null
+          || level == IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED
+          || level == IsolationLevel.UNRECOGNIZED;
+    }
+
+    private static boolean isUnspecified(ReadLockMode mode) {
+      return mode == null
+          || mode == ReadLockMode.READ_LOCK_MODE_UNSPECIFIED
+          || mode == ReadLockMode.UNRECOGNIZED;
+    }
+
+    /**
+     * Determines whether Leader-Routed Read-Your-Writes (LRYW) can be enabled (`routeToLeader =
+     * false`). LRYW is enabled when readLockMode is OPTIMISTIC, or when isolation level is
+     * REPEATABLE_READ and readLockMode is unspecified.
+     */
+    private static boolean canEnableLRYW(TransactionMode mode) {
+      return mode.readLockMode == ReadLockMode.OPTIMISTIC
+          || (mode.isolationLevel == IsolationLevel.REPEATABLE_READ
+              && isUnspecified(mode.readLockMode));
     }
 
     @Override
@@ -247,7 +335,7 @@ protected boolean isReadOnly() {
 
     @Override
     protected boolean isRouteToLeader() {
-      return true;
+      return routeToLeader;
     }
 
     private void increaseAsyncOperations() {