Skip to content

Commit ae4280e

Browse files
shobhitsgshobhitsg
committed
feat(spanner): implement transaction routing logic based on database metadata isolation levels and lock modes
1 parent d586d07 commit ae4280e

20 files changed

Lines changed: 852 additions & 166 deletions

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

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.spanner;
18+
19+
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
20+
import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
21+
import java.util.Objects;
22+
23+
/**
24+
* Internal container for dynamic database-level defaults queried from
25+
* INFORMATION_SCHEMA.DATABASE_OPTIONS. Holds the database dialect, default transaction isolation
26+
* level, and default read lock mode.
27+
*/
28+
final class DatabaseMetadata {
29+
private final Dialect dialect;
30+
private final IsolationLevel isolationLevel;
31+
private final ReadLockMode readLockMode;
32+
33+
DatabaseMetadata(Dialect dialect, IsolationLevel isolationLevel, ReadLockMode readLockMode) {
34+
this.dialect = Objects.requireNonNull(dialect);
35+
this.isolationLevel = Objects.requireNonNull(isolationLevel);
36+
this.readLockMode = Objects.requireNonNull(readLockMode);
37+
}
38+
39+
Dialect getDialect() {
40+
return dialect;
41+
}
42+
43+
IsolationLevel getIsolationLevel() {
44+
return isolationLevel;
45+
}
46+
47+
ReadLockMode getReadLockMode() {
48+
return readLockMode;
49+
}
50+
51+
@Override
52+
public boolean equals(Object o) {
53+
if (this == o) {
54+
return true;
55+
}
56+
if (o == null || getClass() != o.getClass()) {
57+
return false;
58+
}
59+
DatabaseMetadata that = (DatabaseMetadata) o;
60+
return dialect == that.dialect
61+
&& isolationLevel == that.isolationLevel
62+
&& readLockMode == that.readLockMode;
63+
}
64+
65+
@Override
66+
public int hashCode() {
67+
return Objects.hash(dialect, isolationLevel, readLockMode);
68+
}
69+
70+
@Override
71+
public String toString() {
72+
return String.format(
73+
"DatabaseMetadata{dialect=%s, isolation=%s, lockMode=%s}",
74+
dialect, isolationLevel, readLockMode);
75+
}
76+
}

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

Lines changed: 85 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@
3131
import com.google.common.annotations.VisibleForTesting;
3232
import com.google.common.base.Preconditions;
3333
import com.google.spanner.v1.BatchWriteResponse;
34+
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
35+
import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
3436
import java.time.Clock;
3537
import java.time.Duration;
3638
import java.time.Instant;
@@ -63,14 +65,30 @@ final class MultiplexedSessionDatabaseClient extends AbstractMultiplexedSessionD
6365
*/
6466
private static final int MAX_INITIAL_CREATE_SESSION_ATTEMPTS = 10;
6567

68+
/**
69+
* Statement used to query database-level default options from
70+
* INFORMATION_SCHEMA.DATABASE_OPTIONS. This retrieves 'default_transaction_isolation',
71+
* 'default_read_lock_mode', and 'database_dialect' so that the client can correctly configure
72+
* transaction routing (e.g. Leader-Aware Routing) and isolation level behaviors without relying
73+
* solely on client-side hardcoded defaults.
74+
*/
6675
@VisibleForTesting
67-
static final Statement DETERMINE_DIALECT_STATEMENT =
76+
static final Statement DETERMINE_METADATA_STATEMENT =
6877
Statement.newBuilder(
69-
"select option_value "
70-
+ "from information_schema.database_options "
71-
+ "where option_name='database_dialect'")
78+
"SELECT OPTION_NAME, OPTION_VALUE "
79+
+ "FROM INFORMATION_SCHEMA.DATABASE_OPTIONS "
80+
+ "WHERE OPTION_NAME IN ('default_transaction_isolation', "
81+
+ "'default_read_lock_mode', 'database_dialect')")
7282
.build();
7383

84+
static final String OPTION_DATABASE_DIALECT = "database_dialect";
85+
static final String OPTION_DEFAULT_TRANSACTION_ISOLATION = "default_transaction_isolation";
86+
static final String OPTION_DEFAULT_READ_LOCK_MODE = "default_read_lock_mode";
87+
88+
static final String ISOLATION_LEVEL_REPEATABLE_READ = "repeatable read";
89+
static final String READ_LOCK_MODE_OPTIMISTIC = "optimistic";
90+
static final String READ_LOCK_MODE_PESSIMISTIC = "pessimistic";
91+
7492
/**
7593
* Represents a single transaction on a multiplexed session. This can be both a single-use or
7694
* 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) {
274292
// only start the maintainer if we actually managed to create a session in the first
275293
// place.
276294
maintainer.start();
277-
if (sessionClient
278-
.getSpanner()
279-
.getOptions()
280-
.getSessionPoolOptions()
281-
.isAutoDetectDialect()) {
282-
MAINTAINER_SERVICE.submit(() -> getDialect());
283-
}
295+
MAINTAINER_SERVICE.submit(
296+
() -> session.getSessionReference().setDatabaseMetadata(getDatabaseMetadata()));
284297
}
285298

286299
@Override
@@ -371,6 +384,12 @@ AtomicLong getNumSessionsReleased() {
371384
return this.numSessionsReleased;
372385
}
373386

387+
@VisibleForTesting
388+
void resetAcquiredAndReleasedCounts() {
389+
this.numSessionsAcquired.set(0L);
390+
this.numSessionsReleased.set(0L);
391+
}
392+
374393
void close() {
375394
boolean releaseChannelUsage = false;
376395
synchronized (this) {
@@ -473,32 +492,71 @@ private int getSingleUseChannelHint() {
473492
}
474493
}
475494

476-
private final AbstractLazyInitializer<Dialect> dialectSupplier =
477-
new AbstractLazyInitializer<Dialect>() {
495+
private static IsolationLevel parseIsolationLevel(String value) {
496+
return ISOLATION_LEVEL_REPEATABLE_READ.equalsIgnoreCase(value)
497+
? IsolationLevel.REPEATABLE_READ
498+
: IsolationLevel.SERIALIZABLE;
499+
}
500+
501+
private static ReadLockMode parseReadLockMode(String value) {
502+
if (READ_LOCK_MODE_OPTIMISTIC.equalsIgnoreCase(value)) {
503+
return ReadLockMode.OPTIMISTIC;
504+
} else if (READ_LOCK_MODE_PESSIMISTIC.equalsIgnoreCase(value)) {
505+
return ReadLockMode.PESSIMISTIC;
506+
}
507+
return ReadLockMode.READ_LOCK_MODE_UNSPECIFIED;
508+
}
509+
510+
/**
511+
* Lazily initializes and caches {@link DatabaseMetadata} (dialect, default isolation level, and
512+
* read lock mode). Introspects the database options once and attaches the resolved metadata to
513+
* the current multiplexed {@link SessionReference} so subsequent transactions can resolve their
514+
* effective modes.
515+
*/
516+
private final AbstractLazyInitializer<DatabaseMetadata> metadataSupplier =
517+
new AbstractLazyInitializer<DatabaseMetadata>() {
478518
@Override
479-
protected Dialect initialize() {
480-
try (ResultSet dialectResultSet = singleUse().executeQuery(DETERMINE_DIALECT_STATEMENT)) {
481-
if (dialectResultSet.next()) {
482-
return Dialect.fromName(dialectResultSet.getString(0));
519+
protected DatabaseMetadata initialize() {
520+
Dialect dialect = Dialect.GOOGLE_STANDARD_SQL;
521+
IsolationLevel isolationLevel = IsolationLevel.SERIALIZABLE;
522+
ReadLockMode readLockMode = ReadLockMode.READ_LOCK_MODE_UNSPECIFIED;
523+
524+
numSessionsAcquired.decrementAndGet();
525+
try (ResultSet resultSet = singleUse().executeQuery(DETERMINE_METADATA_STATEMENT)) {
526+
while (resultSet.next()) {
527+
String name = resultSet.getString(0);
528+
String value = resultSet.getString(1);
529+
if (OPTION_DATABASE_DIALECT.equalsIgnoreCase(name)) {
530+
dialect = Dialect.fromName(value);
531+
} else if (OPTION_DEFAULT_TRANSACTION_ISOLATION.equalsIgnoreCase(name)) {
532+
isolationLevel = parseIsolationLevel(value);
533+
} else if (OPTION_DEFAULT_READ_LOCK_MODE.equalsIgnoreCase(name)) {
534+
readLockMode = parseReadLockMode(value);
535+
}
483536
}
537+
} finally {
538+
numSessionsReleased.decrementAndGet();
484539
}
485-
// This should not really happen, but it is the safest fallback value.
486-
return Dialect.GOOGLE_STANDARD_SQL;
540+
return new DatabaseMetadata(dialect, isolationLevel, readLockMode);
487541
}
488542
};
489543

490-
@Override
491-
public Dialect getDialect() {
544+
DatabaseMetadata getDatabaseMetadata() {
492545
try {
493-
return dialectSupplier.get();
546+
return metadataSupplier.get();
494547
} catch (Exception exception) {
495548
throw SpannerExceptionFactory.asSpannerException(exception);
496549
}
497550
}
498551

552+
@Override
553+
public Dialect getDialect() {
554+
return getDatabaseMetadata().getDialect();
555+
}
556+
499557
Future<Dialect> getDialectAsync() {
500558
try {
501-
return MAINTAINER_SERVICE.submit(dialectSupplier::get);
559+
return MAINTAINER_SERVICE.submit(() -> getDialect());
502560
} catch (Exception exception) {
503561
throw SpannerExceptionFactory.asSpannerException(exception);
504562
}
@@ -659,8 +717,10 @@ void maintain() {
659717
new SessionConsumer() {
660718
@Override
661719
public void onSessionReady(SessionImpl session) {
662-
multiplexedSessionReference.set(
663-
ApiFutures.immediateFuture(session.getSessionReference()));
720+
SessionReference sessionRef = session.getSessionReference();
721+
multiplexedSessionReference.set(ApiFutures.immediateFuture(sessionRef));
722+
MAINTAINER_SERVICE.submit(
723+
() -> sessionRef.setDatabaseMetadata(getDatabaseMetadata()));
664724
expirationDate.set(
665725
clock
666726
.instant()

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ class SessionReference {
3838
private volatile Instant lastUseTime;
3939
@Nullable private final Instant createTime;
4040
private final boolean isMultiplexed;
41+
private volatile DatabaseMetadata databaseMetadata;
4142

4243
SessionReference(String name, @Nullable String databaseRole, Map<SpannerRpc.Option, ?> options) {
4344
this.options = options;
@@ -92,6 +93,15 @@ boolean getIsMultiplexed() {
9293
return isMultiplexed;
9394
}
9495

96+
@Nullable
97+
DatabaseMetadata getDatabaseMetadata() {
98+
return databaseMetadata;
99+
}
100+
101+
void setDatabaseMetadata(DatabaseMetadata databaseMetadata) {
102+
this.databaseMetadata = databaseMetadata;
103+
}
104+
95105
void markUsed(Instant instant) {
96106
lastUseTime = instant;
97107
}

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

Lines changed: 89 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,8 @@
5454
import com.google.spanner.v1.RollbackRequest;
5555
import com.google.spanner.v1.Transaction;
5656
import com.google.spanner.v1.TransactionOptions;
57+
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
58+
import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
5759
import com.google.spanner.v1.TransactionSelector;
5860
import java.util.ArrayList;
5961
import java.util.List;
@@ -223,8 +225,19 @@ public void removeListener(Runnable listener) {
223225
private CommitResponse commitResponse;
224226
private final Clock clock;
225227

228+
private final boolean routeToLeader;
226229
private final Map<SpannerRpc.Option, ?> channelHint;
227230

231+
private static final class TransactionMode {
232+
private final IsolationLevel isolationLevel;
233+
private final ReadLockMode readLockMode;
234+
235+
TransactionMode(IsolationLevel isolationLevel, ReadLockMode readLockMode) {
236+
this.isolationLevel = isolationLevel;
237+
this.readLockMode = readLockMode;
238+
}
239+
}
240+
228241
private TransactionContextImpl(Builder builder) {
229242
super(builder);
230243
this.transactionId = builder.transactionId;
@@ -238,6 +251,81 @@ private TransactionContextImpl(Builder builder) {
238251
ThreadLocalRandom.current().nextLong(Long.MAX_VALUE),
239252
session.getSpanner().getOptions().isGrpcGcpExtensionEnabled());
240253
this.previousTransactionId = builder.previousTransactionId;
254+
255+
TransactionMode effectiveMode = resolveEffectiveMode(this.options);
256+
this.routeToLeader = !canEnableLRYW(effectiveMode);
257+
}
258+
259+
/**
260+
* Resolves the effective isolation level and read lock mode using the following precedence: 1.
261+
* Call-site options explicitly passed to the transaction (`callSite`). 2. Client-side static
262+
* default transaction options configured on `SpannerOptions`. 3. Database-level defaults
263+
* queried from `INFORMATION_SCHEMA.DATABASE_OPTIONS` (`dbDefaults`). 4. Hardcoded spanner
264+
* defaults (`SERIALIZABLE` isolation level).
265+
*/
266+
private TransactionMode resolveEffectiveMode(Options callSite) {
267+
IsolationLevel isolationLevel = callSite.isolationLevel();
268+
ReadLockMode readLockMode = callSite.readLockMode();
269+
270+
TransactionOptions defaultTxOptions = null;
271+
if (session.getSpanner() != null && session.getSpanner().getOptions() != null) {
272+
defaultTxOptions = session.getSpanner().getOptions().getDefaultTransactionOptions();
273+
}
274+
if (defaultTxOptions != null) {
275+
if (isUnspecified(isolationLevel)) {
276+
isolationLevel = defaultTxOptions.getIsolationLevel();
277+
}
278+
if (isUnspecified(readLockMode) && defaultTxOptions.hasReadWrite()) {
279+
readLockMode = defaultTxOptions.getReadWrite().getReadLockMode();
280+
}
281+
}
282+
283+
DatabaseMetadata dbDefaults = null;
284+
if (session.getSessionReference() != null) {
285+
dbDefaults = session.getSessionReference().getDatabaseMetadata();
286+
}
287+
if (dbDefaults != null) {
288+
if (isUnspecified(isolationLevel)) {
289+
isolationLevel = dbDefaults.getIsolationLevel();
290+
}
291+
if (isUnspecified(readLockMode)) {
292+
readLockMode = dbDefaults.getReadLockMode();
293+
}
294+
}
295+
296+
if (isUnspecified(isolationLevel)) {
297+
isolationLevel = IsolationLevel.SERIALIZABLE;
298+
}
299+
// For REPEATABLE_READ, keep lock mode unspecified/null when not explicitly set so that
300+
// canEnableLRYW evaluates to true for Leader-Routed Read-Your-Writes.
301+
if (isUnspecified(readLockMode) && isolationLevel != IsolationLevel.REPEATABLE_READ) {
302+
readLockMode = ReadLockMode.PESSIMISTIC;
303+
}
304+
305+
return new TransactionMode(isolationLevel, readLockMode);
306+
}
307+
308+
private static boolean isUnspecified(IsolationLevel level) {
309+
return level == null
310+
|| level == IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED
311+
|| level == IsolationLevel.UNRECOGNIZED;
312+
}
313+
314+
private static boolean isUnspecified(ReadLockMode mode) {
315+
return mode == null
316+
|| mode == ReadLockMode.READ_LOCK_MODE_UNSPECIFIED
317+
|| mode == ReadLockMode.UNRECOGNIZED;
318+
}
319+
320+
/**
321+
* Determines whether Leader-Routed Read-Your-Writes (LRYW) can be enabled (`routeToLeader =
322+
* false`). LRYW is enabled when readLockMode is OPTIMISTIC, or when isolation level is
323+
* REPEATABLE_READ and readLockMode is unspecified.
324+
*/
325+
private static boolean canEnableLRYW(TransactionMode mode) {
326+
return mode.readLockMode == ReadLockMode.OPTIMISTIC
327+
|| (mode.isolationLevel == IsolationLevel.REPEATABLE_READ
328+
&& isUnspecified(mode.readLockMode));
241329
}
242330

243331
@Override
@@ -247,7 +335,7 @@ protected boolean isReadOnly() {
247335

248336
@Override
249337
protected boolean isRouteToLeader() {
250-
return true;
338+
return routeToLeader;
251339
}
252340

253341
private void increaseAsyncOperations() {

0 commit comments

Comments
 (0)