Skip to content

refactor(spanner): buffer chunks and add non-blocking isDataAvailable to GrpcValueIterator #14699

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

Open
olavloite wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(spanner): buffer chunks and add non-blocking isDataAvailable to GrpcValueIterator #14699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
17 changes: 11 additions & 6 deletions java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import javax.annotation.Nullable;
import org.jspecify.annotations.Nullable;

@VisibleForTesting
class GrpcResultSet extends AbstractResultSet<List<Object>>
Expand All @@ -39,9 +39,8 @@ class GrpcResultSet extends AbstractResultSet<List<Object>>
private ResultSetMetadata metadata;
private GrpcStruct currRow;
private List<Object> rowData;
private SpannerException error;
private ResultSetStats statistics;
private boolean closed;
private volatile boolean closed;

GrpcResultSet(CloseableIterator<PartialResultSet> iterator, Listener listener) {
this(iterator, listener, DecodeMode.DIRECT);
Expand Down Expand Up @@ -75,9 +74,6 @@ protected GrpcStruct currRow() {

@Override
public boolean next() throws SpannerException {
if (error != null) {
throw asSpannerException(error);
}
try {
if (currRow == null) {
metadata = iterator.getMetadata();
Expand Down Expand Up @@ -132,6 +128,15 @@ public boolean initiateStreaming(AsyncResultSet.StreamMessageListener streamMess
return iterator.initiateStreaming(streamMessageListener);
}

@Override
@InternalApi
public boolean isDataAvailable() {
// If the result set is already closed, return true so callers (such as async stream
// consumers or event loops) do not block or wait indefinitely expecting future data,
// and instead proceed immediately to observe EOF or fast-fail.
return closed || iterator.isDataAvailable();
}

@Override
public void close() {
synchronized (this) {
Expand Down
Loading
Loading