diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ForwardingResultSet.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ForwardingResultSet.java index 3c4883e65862..908bdf9058e4 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ForwardingResultSet.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/ForwardingResultSet.java @@ -110,4 +110,10 @@ public ResultSetMetadata getMetadata() { public boolean initiateStreaming(AsyncResultSet.StreamMessageListener streamMessageListener) { return StreamingUtil.initiateStreaming(delegate.get(), streamMessageListener); } + + @Override + @InternalApi + public boolean isDataAvailable() { + return StreamingUtil.isDataAvailable(delegate.get()); + } } diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingResultSet.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingResultSet.java index 47b10d852c64..cd48885aae1d 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingResultSet.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingResultSet.java @@ -28,4 +28,22 @@ interface StreamingResultSet extends ResultSet { */ @InternalApi boolean initiateStreaming(AsyncResultSet.StreamMessageListener streamMessageListener); + + /** + * Returns true if data (a chunk, row, EOF, or error) is available to read immediately without + * blocking the calling thread on network I/O. + * + *

The default implementation returns {@code true} to maintain backward compatibility with + * synchronous implementations, conservatively indicating that data may be available rather than + * causing the calling thread to wait indefinitely. If the result set has already been closed, + * this also returns {@code true} because subsequent read operations will complete or fail fast + * without blocking. + * + * @return true if data (a chunk, row, EOF, or error) is available without blocking on network + * I/O; false otherwise. + */ + @InternalApi + default boolean isDataAvailable() { + return true; + } } diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingUtil.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingUtil.java index 54496d39f965..d6c932147589 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingUtil.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/StreamingUtil.java @@ -16,6 +16,8 @@ package com.google.cloud.spanner; +import com.google.common.base.Preconditions; + final class StreamingUtil { private StreamingUtil() {} @@ -27,4 +29,12 @@ static boolean initiateStreaming( } return false; } + + static boolean isDataAvailable(ResultSet resultSet) { + Preconditions.checkNotNull(resultSet); + if (resultSet instanceof StreamingResultSet) { + return ((StreamingResultSet) resultSet).isDataAvailable(); + } + return true; + } } diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ForwardingResultSetTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ForwardingResultSetTest.java new file mode 100644 index 000000000000..f5e742fad1b2 --- /dev/null +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ForwardingResultSetTest.java @@ -0,0 +1,176 @@ +/* + * 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 org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.base.Supplier; +import com.google.protobuf.Value; +import com.google.spanner.v1.ResultSetMetadata; +import com.google.spanner.v1.ResultSetStats; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ForwardingResultSetTest { + + @Test + public void testDelegation() { + StreamingResultSet delegate = mock(StreamingResultSet.class); + when(delegate.next()).thenReturn(true, false); + ResultSetStats stats = ResultSetStats.getDefaultInstance(); + ResultSetMetadata metadata = ResultSetMetadata.getDefaultInstance(); + when(delegate.getStats()).thenReturn(stats); + when(delegate.getMetadata()).thenReturn(metadata); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertTrue(forwardingResultSet.next()); + assertFalse(forwardingResultSet.next()); + assertEquals(stats, forwardingResultSet.getStats()); + assertEquals(metadata, forwardingResultSet.getMetadata()); + assertEquals(delegate, forwardingResultSet.getDelegate()); + } + + verify(delegate).close(); + } + + @Test + public void testInitiateStreamingDelegation() { + StreamingResultSet delegate = mock(StreamingResultSet.class); + AsyncResultSet.StreamMessageListener listener = + mock(AsyncResultSet.StreamMessageListener.class); + when(delegate.initiateStreaming(listener)).thenReturn(true); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertTrue(forwardingResultSet.initiateStreaming(listener)); + verify(delegate).initiateStreaming(listener); + } + } + + @Test + public void testIsDataAvailableDelegation() { + StreamingResultSet delegate = mock(StreamingResultSet.class); + when(delegate.isDataAvailable()).thenReturn(false, true); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertFalse(forwardingResultSet.isDataAvailable()); + assertTrue(forwardingResultSet.isDataAvailable()); + } + } + + @Test + public void testIsDataAvailableWhenDelegateIsNotStreamingResultSet() { + ResultSet delegate = mock(ResultSet.class); + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertTrue(forwardingResultSet.isDataAvailable()); + } + } + + @Test + public void testReplaceDelegate() { + ResultSet firstDelegate = mock(ResultSet.class); + ResultSet secondDelegate = mock(ResultSet.class); + when(firstDelegate.next()).thenReturn(true); + when(secondDelegate.next()).thenReturn(false); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(firstDelegate)) { + assertTrue(forwardingResultSet.next()); + + forwardingResultSet.replaceDelegate(secondDelegate); + assertFalse(forwardingResultSet.next()); + assertEquals(secondDelegate, forwardingResultSet.getDelegate()); + } + } + + @Test + public void testSupplierConstructor() { + ResultSet delegate = mock(ResultSet.class); + when(delegate.next()).thenReturn(true); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(() -> delegate)) { + assertTrue(forwardingResultSet.next()); + assertEquals(delegate, forwardingResultSet.getDelegate()); + } + } + + @Test + public void testGetCurrentRowAsStruct() { + ResultSet delegate = mock(ResultSet.class); + Struct struct = mock(Struct.class); + when(delegate.getCurrentRowAsStruct()).thenReturn(struct); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertEquals(struct, forwardingResultSet.getCurrentRowAsStruct()); + } + } + + @Test + public void testProtobufResultSetDelegationWhenSupported() { + ProtobufResultSet delegate = mock(ProtobufResultSet.class); + Value value = Value.newBuilder().setStringValue("test").build(); + when(delegate.canGetProtobufValue(0)).thenReturn(true); + when(delegate.getProtobufValue(0)).thenReturn(value); + + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertTrue(forwardingResultSet.canGetProtobufValue(0)); + assertEquals(value, forwardingResultSet.getProtobufValue(0)); + } + } + + @Test + public void testProtobufResultSetDelegationWhenNotSupported() { + ResultSet delegate = mock(ResultSet.class); + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertFalse(forwardingResultSet.canGetProtobufValue(0)); + IllegalStateException exception = + assertThrows(IllegalStateException.class, () -> forwardingResultSet.getProtobufValue(0)); + assertEquals("The result set does not support protobuf values", exception.getMessage()); + } + } + + @Test + public void testCloseWhenSupplierThrowsException() { + Supplier failingSupplier = + () -> { + throw new RuntimeException("failed to obtain delegate"); + }; + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(failingSupplier)) { + // Exiting the try block invokes close(), testing that it catches and swallows the exception + // cleanly. + } + } + + @Test + public void testConstructorNullCheck() { + assertThrows(NullPointerException.class, () -> new ForwardingResultSet((ResultSet) null)); + } + + @Test + public void testReplaceDelegateNullCheck() { + ResultSet delegate = mock(ResultSet.class); + try (ForwardingResultSet forwardingResultSet = new ForwardingResultSet(delegate)) { + assertThrows(NullPointerException.class, () -> forwardingResultSet.replaceDelegate(null)); + } + } +} diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingUtilTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingUtilTest.java new file mode 100644 index 000000000000..9c6d7a43209d --- /dev/null +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/StreamingUtilTest.java @@ -0,0 +1,119 @@ +/* + * 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 org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class StreamingUtilTest { + + @Test + public void testInitiateStreamingOnStreamingResultSet() { + StreamingResultSet streamingResultSet = mock(StreamingResultSet.class); + AsyncResultSet.StreamMessageListener listener = + mock(AsyncResultSet.StreamMessageListener.class); + when(streamingResultSet.initiateStreaming(listener)).thenReturn(true); + + assertTrue(StreamingUtil.initiateStreaming(streamingResultSet, listener)); + verify(streamingResultSet).initiateStreaming(listener); + } + + @Test + public void testInitiateStreamingOnNonStreamingResultSetReturnsFalse() { + ResultSet nonStreamingResultSet = mock(ResultSet.class); + AsyncResultSet.StreamMessageListener listener = + mock(AsyncResultSet.StreamMessageListener.class); + + assertFalse(StreamingUtil.initiateStreaming(nonStreamingResultSet, listener)); + } + + @Test + public void testIsDataAvailableOnStreamingResultSet() { + StreamingResultSet streamingResultSet = mock(StreamingResultSet.class); + when(streamingResultSet.isDataAvailable()).thenReturn(false, true); + + assertFalse(StreamingUtil.isDataAvailable(streamingResultSet)); + assertTrue(StreamingUtil.isDataAvailable(streamingResultSet)); + } + + @Test + public void testDefaultIsDataAvailableOnStreamingResultSet() { + StreamingResultSet streamingResultSet = mock(StreamingResultSet.class, CALLS_REAL_METHODS); + assertTrue(streamingResultSet.isDataAvailable()); + } + + @Test + public void testIsDataAvailableOnNonStreamingResultSetReturnsTrue() { + ResultSet nonStreamingResultSet = mock(ResultSet.class); + assertTrue(StreamingUtil.isDataAvailable(nonStreamingResultSet)); + } + + @Test + public void testIsDataAvailableOnNestedForwardingResultSet() { + StreamingResultSet underlyingStreamingResultSet = mock(StreamingResultSet.class); + try (ForwardingResultSet innerForwarder = + new ForwardingResultSet(underlyingStreamingResultSet); + ForwardingResultSet outerForwarder = new ForwardingResultSet(innerForwarder)) { + when(underlyingStreamingResultSet.isDataAvailable()).thenReturn(false); + assertFalse(StreamingUtil.isDataAvailable(outerForwarder)); + + when(underlyingStreamingResultSet.isDataAvailable()).thenReturn(true); + assertTrue(StreamingUtil.isDataAvailable(outerForwarder)); + } + } + + @Test + public void testInitiateStreamingOnForwardingResultSet() { + StreamingResultSet underlyingStreamingResultSet = mock(StreamingResultSet.class); + AsyncResultSet.StreamMessageListener listener = + mock(AsyncResultSet.StreamMessageListener.class); + try (ForwardingResultSet forwarder = new ForwardingResultSet(underlyingStreamingResultSet)) { + when(underlyingStreamingResultSet.initiateStreaming(listener)).thenReturn(true); + assertTrue(StreamingUtil.initiateStreaming(forwarder, listener)); + verify(underlyingStreamingResultSet).initiateStreaming(listener); + } + } + + @Test + public void testIsDataAvailableOnForwardingResultSetSubclassOverridingIsDataAvailable() { + ResultSet nonStreamingDelegate = mock(ResultSet.class); + try (ForwardingResultSet customForwarder = + new ForwardingResultSet(nonStreamingDelegate) { + @Override + public boolean isDataAvailable() { + return false; + } + }) { + assertFalse(StreamingUtil.isDataAvailable(customForwarder)); + } + } + + @Test + public void testIsDataAvailableWithNullResultSetThrows() { + assertThrows(NullPointerException.class, () -> StreamingUtil.isDataAvailable(null)); + } +}