perf(spanner): make internal AsyncResultSet row production non-blocking

[olavloite]

Refactors AsyncResultSetImpl to produce rows cooperatively without holding executor threads. Uses isDataAvailable on underlying streaming iterators to avoid blocking on network I/O or full buffers, eliminating thread-per-query pinning and improving throughput under concurrency.

[gemini-code-assist]

Code Review

This pull request introduces a non-blocking isDataAvailable() check across Spanner result set classes to determine if data can be read immediately without blocking on network I/O. It refactors AsyncResultSetImpl to use a state-machine-based scheduling mechanism for producers and callbacks, replacing the previous blocking latch-based synchronization. Additionally, comprehensive unit tests are added to verify these changes. The review feedback highlights that catching Throwable in ProduceRowsRunnable can swallow InterruptedException, and recommends restoring the thread's interrupted status if an InterruptedException is caught.

[olavloite]

/gemini review

[olavloite]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the asynchronous result set implementation (AsyncResultSetImpl) to support non-blocking streaming data checks via a new isDataAvailable() method. It replaces the previous latch-based synchronization mechanism with a state-driven scheduling model for producers and callbacks, and introduces strict null-safety annotations (@NullMarked and @Nullable). The review feedback highlights three critical issues in AsyncResultSetImpl where the nullable fields callback, executor, and result are dereferenced directly without local capture or null checks, which could lead to NullPointerExceptions and violates the newly introduced null-safety rules.

[olavloite]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces a non-blocking data availability check (isDataAvailable()) across various result set and iterator classes in the Google Cloud Spanner library. It also refactors the concurrency model in AsyncResultSetImpl by replacing several CountDownLatches with state flags and explicit scheduling methods to improve thread safety and prevent blocking. The feedback suggests replacing the synchronized blocks in AsyncResultSetImpl with explicit ReentrantLocks to avoid carrier thread pinning when using virtual threads.

[olavloite]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces non-blocking data availability checks (isDataAvailable()) across various result set and iterator implementations, including AsyncResultSetImpl, GrpcStreamIterator, GrpcValueIterator, and ResumableStreamIterator. Additionally, AsyncResultSetImpl is refactored to replace thread-blocking latches with state flags to coordinate producer and consumer threads asynchronously. The review feedback highlights critical race conditions in ResumableStreamIterator.java where gRPC streams could be leaked or started after the iterator is closed, and suggests introducing a closed flag to safely manage the lifecycle of the iterator and its streams.

[gemini-code-assist]

To prevent race conditions and avoid starting new gRPC streams after the iterator has been closed (for example, if a retry is scheduled or in progress when close() is called), we should introduce a private volatile boolean closed; field to track the closed state of the iterator.

  public void close(@Nullable String message) {
    closed = true;
    if (stream != null) {
      stream.close(message);
      span.end();
      stream = null;
    }
  }

References

1. When using lazily initialized resources, ensure that teardown or close methods perform explicit null checks before invoking methods on them to prevent NullPointerException.

[gemini-code-assist]

There is a potential race condition here:

1. startStream is called, which internally calls setStream(stream) to set this.stream = streamIterator.
2. Before startGrpcStreaming() continues, another thread calls close(), which closes the stream and sets this.stream = null.
3. startGrpcStreaming() resumes, sees this.stream == null, and sets this.stream = streamIterator (which is the already-closed stream).

This leaks the closed stream and leaves this.stream non-null, so it will never be closed again. By checking the closed flag, we can safely close the newly created stream if the iterator was closed concurrently, and avoid leaking it.

        if (!closed) {
          stream = streamIterator;
        } else {
          streamIterator.close(null);
        }