Skip to content

fix(spanner): ensure executeQueryAsync is non-blocking #12715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
olavloite merged 2 commits into from
Apr 9, 2026
Merged

fix(spanner): ensure executeQueryAsync is non-blocking #12715

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8974940
fix(spanner): ensure executeQueryAsync is non-blocking
olavloite Apr 8, 2026
5f42200
fix(spanner): only use background executor for read-only tx
olavloite Apr 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions java-spanner/.gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
.vscode
45 changes: 43 additions & 2 deletions ...nner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AbstractReadContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,6 +407,47 @@ TransactionSelector getTransactionSelector() {
return selector;
}

@Override
public ListenableAsyncResultSet readAsync(
String table, KeySet keys, Iterable<String> columns, ReadOption... options) {
Options readOptions = Options.fromReadOptions(options);
final int bufferRows =
readOptions.hasBufferRows()
? readOptions.bufferRows()
: AsyncResultSetImpl.DEFAULT_BUFFER_SIZE;
return new AsyncResultSetImpl(
executorProvider, () -> readInternal(table, null, keys, columns, options), bufferRows);
}

@Override
public ListenableAsyncResultSet readUsingIndexAsync(
String table, String index, KeySet keys, Iterable<String> columns, ReadOption... options) {
Options readOptions = Options.fromReadOptions(options);
final int bufferRows =
readOptions.hasBufferRows()
? readOptions.bufferRows()
: AsyncResultSetImpl.DEFAULT_BUFFER_SIZE;
return new AsyncResultSetImpl(
executorProvider,
() -> readInternal(table, checkNotNull(index), keys, columns, options),
bufferRows);
}

@Override
public ListenableAsyncResultSet executeQueryAsync(Statement statement, QueryOption... options) {
Options readOptions = Options.fromQueryOptions(options);
final int bufferRows =
readOptions.hasBufferRows()
? readOptions.bufferRows()
: AsyncResultSetImpl.DEFAULT_BUFFER_SIZE;
return new AsyncResultSetImpl(
executorProvider,
() ->
executeQueryInternal(
statement, com.google.spanner.v1.ExecuteSqlRequest.QueryMode.NORMAL, options),
bufferRows);
}

@Override
public Timestamp getReadTimestamp() {
synchronized (txnLock) {
Expand Down Expand Up @@ -679,7 +720,7 @@ public final ResultSet analyzeQuery(Statement statement, QueryAnalyzeMode readCo
}
}

private ResultSet executeQueryInternal(
ResultSet executeQueryInternal(
Statement statement,
com.google.spanner.v1.ExecuteSqlRequest.QueryMode queryMode,
QueryOption... options) {
Expand Down Expand Up @@ -987,7 +1028,7 @@ public void onDone(boolean withBeginTransaction) {
@Override
public void onPrecommitToken(MultiplexedSessionPrecommitToken token) {}

private ResultSet readInternal(
ResultSet readInternal(
String table,
@Nullable String index,
KeySet keys,
Expand Down
16 changes: 14 additions & 2 deletions ...anner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AsyncResultSetImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,9 +157,21 @@ private enum State {

AsyncResultSetImpl(
ExecutorProvider executorProvider, Supplier<ResultSet> delegate, int bufferSize) {
super(delegate);
this(
executorProvider,
Suppliers.memoize(Preconditions.checkNotNull(delegate)),
bufferSize,
true);
}

private AsyncResultSetImpl(
ExecutorProvider executorProvider,
Supplier<ResultSet> memoizedDelegate,
int bufferSize,
boolean dummy) {
super(memoizedDelegate);
this.executorProvider = Preconditions.checkNotNull(executorProvider);
this.delegateResultSet = Preconditions.checkNotNull(delegate);
this.delegateResultSet = memoizedDelegate;
this.service = MoreExecutors.listeningDecorator(executorProvider.getExecutor());
this.buffer = new LinkedBlockingDeque<>(bufferSize);
}
Expand Down
151 changes: 151 additions & 0 deletions ...le-cloud-spanner/src/test/java/com/google/cloud/spanner/AsyncReadOnlyTransactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
/*
* 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 static com.google.cloud.spanner.MockSpannerTestUtil.READ_ONE_KEY_VALUE_STATEMENT;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertTrue;

import com.google.spanner.v1.BeginTransactionRequest;
import com.google.spanner.v1.ExecuteSqlRequest;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

@RunWith(JUnit4.class)
public class AsyncReadOnlyTransactionTest extends AbstractAsyncTransactionTest {

@Test
public void asyncReadOnlyTransactionIsNonBlocking() throws Exception {
// Warm up session pool to avoid CreateSession blocking when server is frozen.
try (ResultSet resultSet = client().singleUse().executeQuery(READ_ONE_KEY_VALUE_STATEMENT)) {
while (resultSet.next()) {}
}
mockSpanner.reset();

try (ReadOnlyTransaction transaction = client().readOnlyTransaction()) {
mockSpanner.freeze();
// Call executeQueryAsync. It should not block even though mock server is
// frozen!
AsyncResultSet rs = transaction.executeQueryAsync(READ_ONE_KEY_VALUE_STATEMENT);

// Verify that no requests have been sent yet.
assertTrue(mockSpanner.getRequestTypes().isEmpty());

// Now register a callback to start the stream.
final CountDownLatch latch = new CountDownLatch(1);
rs.setCallback(
executor,
resultSet -> {
try {
AsyncResultSet.CursorState state;
while ((state = resultSet.tryNext()) == AsyncResultSet.CursorState.OK) {
// consume
}
if (state == AsyncResultSet.CursorState.DONE) {
latch.countDown();
}
return AsyncResultSet.CallbackResponse.CONTINUE;
} catch (Throwable t) {
latch.countDown();
return AsyncResultSet.CallbackResponse.DONE;
}
});

// Unfreeze the mock server so the background thread can proceed.
mockSpanner.unfreeze();

// Wait for the callback to complete.
assertTrue("Timeout waiting for callback", latch.await(10, TimeUnit.SECONDS));

// Verify that requests were sent on the background thread.
// It should contain one BeginTransaction and one ExecuteSql.
assertThat(mockSpanner.getRequestTypes())
.containsExactly(BeginTransactionRequest.class, ExecuteSqlRequest.class);
}
}

@Test
public void testMultipleQueriesOnlyCallsBeginTransactionOnce() throws Exception {
// Warm up session pool to avoid CreateSession blocking when server is frozen.
try (ResultSet resultSet = client().singleUse().executeQuery(READ_ONE_KEY_VALUE_STATEMENT)) {
while (resultSet.next()) {}
}
mockSpanner.reset();

try (ReadOnlyTransaction transaction = client().readOnlyTransaction()) {
mockSpanner.freeze();
// Call executeQueryAsync twice.
AsyncResultSet rs1 = transaction.executeQueryAsync(READ_ONE_KEY_VALUE_STATEMENT);
AsyncResultSet rs2 = transaction.executeQueryAsync(READ_ONE_KEY_VALUE_STATEMENT);

// Verify that no requests have been sent yet.
assertTrue(mockSpanner.getRequestTypes().isEmpty());

// Unfreeze the mock server.
mockSpanner.unfreeze();

// Now register callbacks to start the streams.
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);

rs1.setCallback(
executor,
resultSet -> {
try {
AsyncResultSet.CursorState state;
while ((state = resultSet.tryNext()) == AsyncResultSet.CursorState.OK) {}
if (state == AsyncResultSet.CursorState.DONE) {
latch1.countDown();
}
return AsyncResultSet.CallbackResponse.CONTINUE;
} catch (Throwable t) {
latch1.countDown();
return AsyncResultSet.CallbackResponse.DONE;
}
});

rs2.setCallback(
executor,
resultSet -> {
try {
AsyncResultSet.CursorState state;
while ((state = resultSet.tryNext()) == AsyncResultSet.CursorState.OK) {}
if (state == AsyncResultSet.CursorState.DONE) {
latch2.countDown();
}
return AsyncResultSet.CallbackResponse.CONTINUE;
} catch (Throwable t) {
latch2.countDown();
return AsyncResultSet.CallbackResponse.DONE;
}
});

// Wait for both callbacks to complete.
assertTrue("Timeout waiting for callback 1", latch1.await(10, TimeUnit.SECONDS));
assertTrue("Timeout waiting for callback 2", latch2.await(10, TimeUnit.SECONDS));

// Verify that requests were sent.
// It should contain one BeginTransaction and two ExecuteSql.
assertThat(mockSpanner.getRequestTypes())
.containsExactly(
BeginTransactionRequest.class, ExecuteSqlRequest.class, ExecuteSqlRequest.class);
}
}
}
Loading