Skip to content

Commit 8464953

Browse files
rayudu3745rayudu3745
committed
fix: use database default isolation level when none is specified
1 parent ec11625 commit 8464953

10 files changed

Lines changed: 495 additions & 27 deletions

File tree

‎src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,6 @@
7676
import com.google.common.util.concurrent.MoreExecutors;
7777
import com.google.spanner.admin.database.v1.InstanceName;
7878
import com.google.spanner.v1.DatabaseName;
79-
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
8079
import java.io.DataOutputStream;
8180
import java.io.EOFException;
8281
import java.io.IOException;
@@ -273,13 +272,6 @@ public void connectToSpanner(
273272
spannerConnection.close();
274273
throw e;
275274
}
276-
// Override the 'unspecified' isolation level to prevent errors in drivers that query this
277-
// setting at startup, and expect one of the supported PostgreSQL isolation levels.
278-
// TODO: Once supported: Query the database for the default isolation level.
279-
if (spannerConnection.getDefaultIsolationLevel()
280-
== IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED) {
281-
spannerConnection.setDefaultIsolationLevel(IsolationLevel.SERIALIZABLE);
282-
}
283275
spannerConnection.setSavepointSupport(SavepointSupport.ENABLED);
284276
this.spannerConnection = spannerConnection;
285277
this.databaseId = connectionOptions.getDatabaseId();

‎src/main/java/com/google/cloud/spanner/pgadapter/session/PGSetting.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ static Integer parseInteger(String value) {
214214
? null
215215
: ImmutableList.copyOf(
216216
Arrays.stream(enumVals)
217-
.map(v -> v.toUpperCase(Locale.ENGLISH))
217+
.map(v -> unquote(v).toUpperCase(Locale.ENGLISH))
218218
.collect(Collectors.toList()));
219219
this.setting = setting;
220220
this.unit = unit;
@@ -249,6 +249,13 @@ PGSetting copy() {
249249
pendingRestart);
250250
}
251251

252+
static String unquote(String value) {
253+
if (value != null && value.length() >= 2 && value.startsWith("\"") && value.endsWith("\"")) {
254+
return value.substring(1, value.length() - 1);
255+
}
256+
return value;
257+
}
258+
252259
static Context tryParseContext(String value) {
253260
try {
254261
return Strings.isNullOrEmpty(value)

‎src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.google.common.collect.ImmutableMap;
3131
import com.google.common.collect.ImmutableSet;
3232
import com.google.common.collect.Sets;
33+
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
3334
import java.time.Duration;
3435
import java.time.ZoneId;
3536
import java.util.ArrayList;
@@ -93,6 +94,10 @@ public class SessionState {
9394
/** localSettings are the modified local settings during a transaction. */
9495
private Map<String, PGSetting> localSettings;
9596

97+
private IsolationLevel initialIsolationLevel = IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED;
98+
private IsolationLevel defaultIsolationLevel = IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED;
99+
private IsolationLevel transactionDefaultIsolationLevel;
100+
96101
public SessionState(OptionsMetadata options) {
97102
this(ImmutableMap.of(), options);
98103
}
@@ -124,6 +129,83 @@ public SessionState(OptionsMetadata options) {
124129
initCopySettings(this.settings);
125130
}
126131

132+
public void setInitialIsolationLevel(@Nullable IsolationLevel initialIsolationLevel) {
133+
this.initialIsolationLevel =
134+
MoreObjects.firstNonNull(initialIsolationLevel, IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED);
135+
this.defaultIsolationLevel = this.initialIsolationLevel;
136+
this.transactionDefaultIsolationLevel = null;
137+
String pgIsolationLevel = toPgIsolationLevel(this.initialIsolationLevel);
138+
setConnectionStartupValue(null, "default_transaction_isolation", pgIsolationLevel);
139+
setConnectionStartupValue(null, "transaction_isolation", pgIsolationLevel);
140+
}
141+
142+
public IsolationLevel getInitialIsolationLevel() {
143+
return this.initialIsolationLevel;
144+
}
145+
146+
public IsolationLevel getDefaultIsolationLevel() {
147+
return this.transactionDefaultIsolationLevel != null
148+
? this.transactionDefaultIsolationLevel
149+
: this.defaultIsolationLevel;
150+
}
151+
152+
public void setDefaultIsolationLevel(
153+
@Nullable IsolationLevel isolationLevel, boolean inTransaction) {
154+
IsolationLevel effectiveLevel =
155+
isolationLevel == null || isolationLevel == IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED
156+
? this.initialIsolationLevel
157+
: isolationLevel;
158+
this.transactionDefaultIsolationLevel = effectiveLevel;
159+
String pgIsolationLevel = toPgIsolationLevel(effectiveLevel);
160+
String currentTxIsolation =
161+
inTransaction ? get(null, "transaction_isolation").getSetting() : null;
162+
set(null, "default_transaction_isolation", pgIsolationLevel);
163+
set(null, "transaction_isolation", pgIsolationLevel);
164+
if (inTransaction && !Objects.equals(currentTxIsolation, pgIsolationLevel)) {
165+
setLocal(null, "transaction_isolation", currentTxIsolation);
166+
}
167+
}
168+
169+
public static IsolationLevel toSpannerIsolationLevel(@Nullable String value) {
170+
if (value == null) {
171+
return IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED;
172+
}
173+
String unquoted = PGSetting.unquote(value.trim());
174+
if (unquoted.length() >= 2 && unquoted.startsWith("'") && unquoted.endsWith("'")) {
175+
unquoted = unquoted.substring(1, unquoted.length() - 1);
176+
}
177+
if ("default".equalsIgnoreCase(unquoted)) {
178+
return IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED;
179+
}
180+
if ("repeatable read".equalsIgnoreCase(unquoted)
181+
|| "repeatable_read".equalsIgnoreCase(unquoted)) {
182+
return IsolationLevel.REPEATABLE_READ;
183+
}
184+
if ("serializable".equalsIgnoreCase(unquoted)) {
185+
return IsolationLevel.SERIALIZABLE;
186+
}
187+
throw PGSetting.invalidEnumError(
188+
"default_transaction_isolation",
189+
value,
190+
new String[] {"serializable", "\"repeatable read\""});
191+
}
192+
193+
public static String toPgIsolationLevel(@Nullable IsolationLevel isolationLevel) {
194+
if (isolationLevel == IsolationLevel.REPEATABLE_READ) {
195+
return "repeatable read";
196+
}
197+
return "serializable";
198+
}
199+
200+
public static String toPgIsolationLevel(@Nullable String spannerIsolationLevel) {
201+
if (spannerIsolationLevel != null
202+
&& ("REPEATABLE_READ".equalsIgnoreCase(spannerIsolationLevel)
203+
|| "REPEATABLE READ".equalsIgnoreCase(spannerIsolationLevel))) {
204+
return "repeatable read";
205+
}
206+
return "serializable";
207+
}
208+
127209
/**
128210
* These fields contain cached values for settings that are used often and/or expensive to
129211
* calculate.
@@ -410,12 +492,18 @@ public List<PGSetting> getAll() {
410492

411493
/** Resets all values to their 'reset' value. */
412494
public void resetAll() {
495+
resetAll(false);
496+
}
497+
498+
/** Resets all values to their 'reset' value. */
499+
public void resetAll(boolean inTransaction) {
413500
for (PGSetting setting : getAll()) {
414501
if (setting.isSettable(Context.SUPERUSER)
415502
&& !Objects.equals(setting.getSetting(), setting.getResetVal())) {
416503
set(setting.getExtension(), setting.getName(), setting.getResetVal());
417504
}
418505
}
506+
setDefaultIsolationLevel(this.initialIsolationLevel, inTransaction);
419507
}
420508

421509
static SpannerException unknownParamError(String key) {
@@ -434,6 +522,10 @@ public void commit() {
434522
settings.put(toKey(setting.getExtension(), setting.getName()), setting);
435523
}
436524
}
525+
if (this.transactionDefaultIsolationLevel != null) {
526+
this.defaultIsolationLevel = this.transactionDefaultIsolationLevel;
527+
this.transactionDefaultIsolationLevel = null;
528+
}
437529
// Dropping the local settings can uncover a different value, so the cache must go. Promoting
438530
// the transaction settings cannot: internalGet(..) returns the same PGSetting instance either
439531
// way. The distinction matters, as commit() runs for every statement in autocommit mode.
@@ -451,6 +543,7 @@ public void rollback() {
451543
if (localSettings != null || transactionSettings != null) {
452544
invalidateCache();
453545
}
546+
this.transactionDefaultIsolationLevel = null;
454547
this.localSettings = null;
455548
this.transactionSettings = null;
456549
}

‎src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java‎

Lines changed: 90 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@
7070
import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
7171
import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
7272
import com.google.common.annotations.VisibleForTesting;
73+
import com.google.common.base.MoreObjects;
7374
import com.google.common.base.Preconditions;
7475
import com.google.common.base.Stopwatch;
7576
import com.google.common.base.Suppliers;
@@ -82,6 +83,7 @@
8283
import com.google.common.util.concurrent.ListeningExecutorService;
8384
import com.google.common.util.concurrent.MoreExecutors;
8485
import com.google.common.util.concurrent.SettableFuture;
86+
import com.google.spanner.v1.TransactionOptions.IsolationLevel;
8587
import io.opentelemetry.api.common.AttributeKey;
8688
import io.opentelemetry.api.common.Attributes;
8789
import io.opentelemetry.api.trace.Span;
@@ -535,6 +537,9 @@ private StatementResult executeOnSpannerWithLogging(Statement statement) {
535537
() -> String.format("Statement: %s", statement.getSql())));
536538
Stopwatch stopwatch = Stopwatch.createStarted();
537539
StatementResult result = spannerConnection.execute(statement);
540+
if (parsedStatement.getType() == StatementType.CLIENT_SIDE) {
541+
syncSessionStateWithClientSideStatement();
542+
}
538543
Duration executionDuration = stopwatch.elapsed();
539544
metrics.recordClientLibLatency(executionDuration.toMillis(), metricAttributes);
540545
logger.log(
@@ -564,6 +569,52 @@ private StatementResult executeOnSpannerWithLogging(Statement statement) {
564569
}
565570
}
566571

572+
private void syncSessionStateWithClientSideStatement() {
573+
ClientSideStatementType type = parsedStatement.getClientSideStatementType();
574+
if (type == ClientSideStatementType.SET_DEFAULT_TRANSACTION_ISOLATION
575+
|| (type == ClientSideStatementType.SET_READONLY
576+
&& statement.getSql().toLowerCase(Locale.ENGLISH).contains("isolation"))) {
577+
if (spannerConnection.getDefaultIsolationLevel() != null) {
578+
sessionState.setDefaultIsolationLevel(
579+
spannerConnection.getDefaultIsolationLevel(), spannerConnection.isInTransaction());
580+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
581+
}
582+
} else if (type == ClientSideStatementType.BEGIN
583+
|| (type == ClientSideStatementType.SET_TRANSACTION_MODE
584+
&& statement.getSql().toLowerCase(Locale.ENGLISH).contains("isolation"))) {
585+
if (spannerConnection.isInTransaction()) {
586+
syncTransactionIsolationLevel();
587+
}
588+
}
589+
}
590+
591+
private void syncTransactionIsolationLevel() {
592+
String isolationLevel = getTransactionIsolationLevelFromConnection();
593+
if (!Objects.equals(
594+
isolationLevel, sessionState.get(null, "transaction_isolation").getSetting())) {
595+
sessionState.setLocal(null, "transaction_isolation", isolationLevel);
596+
}
597+
}
598+
599+
private String getTransactionIsolationLevelFromConnection() {
600+
try {
601+
StatementResult showResult =
602+
spannerConnection.execute(Statement.of("show transaction isolation level"));
603+
if (showResult != null && showResult.getResultSet() != null) {
604+
try (ResultSet rs = showResult.getResultSet()) {
605+
if (rs.next()) {
606+
String level = rs.getString(0);
607+
if (level != null && !"ISOLATION_LEVEL_UNSPECIFIED".equalsIgnoreCase(level)) {
608+
return SessionState.toPgIsolationLevel(level);
609+
}
610+
}
611+
}
612+
}
613+
} catch (Throwable ignore) {
614+
}
615+
return SessionState.toPgIsolationLevel(sessionState.getDefaultIsolationLevel());
616+
}
617+
567618
/**
568619
* Returns true if the given exception is the error that is returned by Cloud Spanner when an
569620
* INFORMATION_SCHEMA query is not executed in a single-use read-only transaction.
@@ -922,6 +973,7 @@ void doExecute() {
922973
this.connectionId = connectionId;
923974
this.closeAllPortals = closeAllPortals;
924975
this.sessionState = new SessionState(optionsMetadata);
976+
this.sessionState.setInitialIsolationLevel(spannerConnection.getDefaultIsolationLevel());
925977
this.pgCatalog =
926978
Suppliers.memoize(
927979
() -> new PgCatalog(BackendConnection.this.sessionState, wellKnownClient.get()));
@@ -1138,6 +1190,8 @@ public void initSessionSetting(String name, String value) {
11381190
ParsedStatement parsedStatement = statementParser.parse(statement);
11391191
if (parsedStatement.getType() == StatementType.CLIENT_SIDE) {
11401192
this.spannerConnection.execute(Statement.of(statement.getSql()));
1193+
this.sessionState.setInitialIsolationLevel(
1194+
this.spannerConnection.getDefaultIsolationLevel());
11411195
continue;
11421196
}
11431197
} catch (Throwable ignore) {
@@ -1166,7 +1220,16 @@ public void initSessionSetting(String name, String value, boolean overwrite) {
11661220
return;
11671221
}
11681222
if (overwrite || this.sessionState.tryGet(key.schema, key.name) == null) {
1169-
this.sessionState.setConnectionStartupValue(key.schema, key.name, value);
1223+
if (key.schema == null && "default_transaction_isolation".equalsIgnoreCase(key.name)) {
1224+
try {
1225+
IsolationLevel level = SessionState.toSpannerIsolationLevel(value);
1226+
this.spannerConnection.setDefaultIsolationLevel(level);
1227+
this.sessionState.setInitialIsolationLevel(level);
1228+
} catch (Throwable ignore) {
1229+
}
1230+
} else {
1231+
this.sessionState.setConnectionStartupValue(key.schema, key.name, value);
1232+
}
11701233
}
11711234
}
11721235

@@ -1234,6 +1297,9 @@ private void flush(boolean isSync) {
12341297
if (canUseBatch) {
12351298
index += executeStatementsInBatch(index);
12361299
} else {
1300+
if (isBegin(index) && !spannerConnection.isInTransaction()) {
1301+
sessionState.commit();
1302+
}
12371303
bufferedStatement.execute();
12381304

12391305
if (isBegin(index)) {
@@ -1246,6 +1312,7 @@ private void flush(boolean isSync) {
12461312
} else {
12471313
sessionState.rollback();
12481314
}
1315+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
12491316
closeAllPortals.run();
12501317
clearCurrentTransaction();
12511318
}
@@ -1269,6 +1336,7 @@ private void flush(boolean isSync) {
12691336
} else if (spannerConnection.isDdlBatchActive()) {
12701337
spannerConnection.abortBatch();
12711338
}
1339+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
12721340
} finally {
12731341
failBufferedStatements(PGExceptionFactory.newQueryCancelledException());
12741342
bufferedStatements.clear();
@@ -1388,13 +1456,31 @@ private void endImplicitTransaction() {
13881456
spannerConnection.setStatementTag(null);
13891457
spannerConnection.commit();
13901458
}
1459+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
13911460
}
13921461
} finally {
13931462
connectionState = ConnectionState.IDLE;
13941463
currentTransactionId = null;
13951464
}
13961465
}
13971466

1467+
static void syncConnectionDefaultIsolationLevel(
1468+
SessionState sessionState, Connection spannerConnection) {
1469+
if (!spannerConnection.isInTransaction()
1470+
&& !spannerConnection.isTransactionStarted()
1471+
&& !spannerConnection.isDdlBatchActive()
1472+
&& !spannerConnection.isDmlBatchActive()) {
1473+
IsolationLevel target = sessionState.getDefaultIsolationLevel();
1474+
IsolationLevel current =
1475+
MoreObjects.firstNonNull(
1476+
spannerConnection.getDefaultIsolationLevel(),
1477+
IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED);
1478+
if (current != target) {
1479+
spannerConnection.setDefaultIsolationLevel(target);
1480+
}
1481+
}
1482+
}
1483+
13981484
/**
13991485
* Prepares the connection for executing a DDL statement. This can include committing the current
14001486
* transaction.
@@ -1428,6 +1514,7 @@ private void prepareExecuteDdl(BufferedStatement<?> bufferedStatement) {
14281514
if (spannerConnection.isInTransaction()) {
14291515
spannerConnection.commit();
14301516
sessionState.commit();
1517+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
14311518
if (transactionMode == TransactionMode.EXPLICIT) {
14321519
// Convert the explicit transaction to a DDL batch.
14331520
transactionMode = TransactionMode.DDL_BATCH;
@@ -1469,6 +1556,7 @@ private void prepareExecuteDdl(BufferedStatement<?> bufferedStatement) {
14691556
if (spannerConnection.isInTransaction()) {
14701557
spannerConnection.commit();
14711558
sessionState.commit();
1559+
syncConnectionDefaultIsolationLevel(sessionState, spannerConnection);
14721560
if (transactionMode == TransactionMode.EXPLICIT) {
14731561
// Convert the explicit transaction to a DDL batch.
14741562
transactionMode = TransactionMode.DDL_BATCH;
@@ -1573,8 +1661,7 @@ SessionStatement getSessionManagementStatement(
15731661
Statement statement, ParsedStatement parsedStatement) {
15741662
if (parsedStatement.getType() == StatementType.UNKNOWN
15751663
|| (parsedStatement.getType() == StatementType.QUERY && isCommand(SHOW, statement.getSql()))
1576-
|| (parsedStatement.getType() == StatementType.CLIENT_SIDE
1577-
&& parsedStatement.getClientSideStatementType() == ClientSideStatementType.RESET_ALL)) {
1664+
|| parsedStatement.getType() == StatementType.CLIENT_SIDE) {
15781665
return SessionStatementParser.parse(parsedStatement, statement.getSql());
15791666
}
15801667
return null;

0 commit comments

Comments
 (0)