Merge branch 'main' into generate-libraries-main

Author: blakeli0

.github/workflows/update_librarian_googleapis.yaml

@@ -33,6 +33,7 @@ jobs:
         repository: googleapis/google-cloud-java
         path: google-cloud-java
         fetch-depth: 0
+        persist-credentials: false
     - name: Check for Open PR
       env:
         GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AbstractReadContext.java

@@ -59,7 +59,10 @@
 import java.util.Collections;
 import java.util.EnumMap;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -321,6 +324,8 @@ static class Builder extends AbstractReadContext.Builder<Builder, MultiUseReadOn
       private TimestampBound bound;
       private Timestamp timestamp;
       private ByteString transactionId;
+      private Options.BeginTransactionOption beginTransactionOption =
+          Options.BeginTransactionOption.EXPLICIT;
 
       private Builder() {}
 
@@ -339,6 +344,11 @@ Builder setTransactionId(ByteString transactionId) {
         return this;
       }
 
+      Builder setBeginTransactionOption(Options.BeginTransactionOption beginTransactionOption) {
+        this.beginTransactionOption = beginTransactionOption;
+        return this;
+      }
+
       @Override
       MultiUseReadOnlyTransaction build() {
         return new MultiUseReadOnlyTransaction(this);
@@ -359,9 +369,15 @@ static Builder newBuilder() {
     @GuardedBy("txnLock")
     private ByteString transactionId;
 
+    @GuardedBy("txnLock")
+    private SettableApiFuture<ByteString> transactionIdFuture;
+
     private final AtomicInteger pendingStarts = new AtomicInteger(0);
 
+    private static final long WAIT_FOR_INLINE_BEGIN_TIMEOUT_MILLIS = 60_000L;
+
     private final Map<SpannerRpc.Option, ?> channelHint;
+    private final Options.BeginTransactionOption beginTransactionOption;
 
     MultiUseReadOnlyTransaction(Builder builder) {
       super(builder);
@@ -386,6 +402,7 @@ static Builder newBuilder() {
               session.getOptions(),
               ThreadLocalRandom.current().nextLong(Long.MAX_VALUE),
               session.getSpanner().getOptions().isGrpcGcpExtensionEnabled());
+      this.beginTransactionOption = builder.beginTransactionOption;
     }
 
     @Override
@@ -398,21 +415,68 @@ protected boolean isRouteToLeader() {
       return false;
     }
 
+    private boolean shouldUseInlinedBegin() {
+      return beginTransactionOption == Options.BeginTransactionOption.INLINE;
+    }
+
     @Override
     void beforeReadOrQuery() {
       super.beforeReadOrQuery();
-      initTransaction();
+      if (shouldUseInlinedBegin()) {
+        // Keep the same nested transaction guard as the explicit BeginTransaction path. This checks
+        // TransactionRunner's thread-local pending state, not the session's active transaction.
+        SessionImpl.throwIfTransactionsPending();
+      } else {
+        initTransaction();
+      }
     }
 
     @Override
     @Nullable
     TransactionSelector getTransactionSelector() {
-      // No need for synchronization: super.readInternal() is always preceded by a check of
-      // "transactionId" that provides a happens-before from initialization, and the value is never
-      // changed afterwards.
-      @SuppressWarnings("GuardedByChecker")
-      TransactionSelector selector = TransactionSelector.newBuilder().setId(transactionId).build();
-      return selector;
+      if (!shouldUseInlinedBegin()) {
+        // No need for synchronization: super.readInternal() is always preceded by a check of
+        // "transactionId" that provides a happens-before from initialization, and the value is
+        // never changed afterwards.
+        @SuppressWarnings("GuardedByChecker")
+        TransactionSelector selector =
+            TransactionSelector.newBuilder().setId(transactionId).build();
+        return selector;
+      }
+
+      ApiFuture<ByteString> futureToWaitFor = null;
+      txnLock.lock();
+      try {
+        if (transactionId != null) {
+          return TransactionSelector.newBuilder().setId(transactionId).build();
+        }
+        if (transactionIdFuture == null) {
+          transactionIdFuture = SettableApiFuture.create();
+          return TransactionSelector.newBuilder()
+              .setBegin(createReadOnlyTransactionOptions())
+              .build();
+        }
+        futureToWaitFor = transactionIdFuture;
+      } finally {
+        txnLock.unlock();
+      }
+
+      try {
+        return TransactionSelector.newBuilder()
+            .setId(futureToWaitFor.get(WAIT_FOR_INLINE_BEGIN_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS))
+            .build();
+      } catch (ExecutionException e) {
+        throw SpannerExceptionFactory.asSpannerException(e.getCause());
+      } catch (TimeoutException e) {
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.DEADLINE_EXCEEDED,
+            "Timeout while waiting for an inlined read-only transaction to be returned by another"
+                + " statement.",
+            e);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw SpannerExceptionFactory.newSpannerExceptionForCancellation(null, e);
+      }
     }
 
     private void decrementPendingStartsAndSignal() {
@@ -503,6 +567,80 @@ public ListenableAsyncResultSet executeQueryAsync(Statement statement, QueryOpti
           bufferRows);
     }
 
+    @Override
+    public void onTransactionMetadata(Transaction transaction, boolean shouldIncludeId) {
+      Timestamp readTimestamp = null;
+      if (transaction.hasReadTimestamp()) {
+        try {
+          readTimestamp = Timestamp.fromProto(transaction.getReadTimestamp());
+        } catch (IllegalArgumentException e) {
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.INTERNAL, "Bad value in transaction.read_timestamp metadata field", e);
+        }
+      }
+      if (shouldIncludeId && transaction.getId().isEmpty()) {
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.FAILED_PRECONDITION, NO_TRANSACTION_RETURNED_MSG);
+      }
+      txnLock.lock();
+      try {
+        if (timestamp == null) {
+          if (readTimestamp == null) {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.INTERNAL, "Missing expected transaction.read_timestamp metadata field");
+          }
+          timestamp = readTimestamp;
+        }
+        if (shouldIncludeId && transactionId == null) {
+          transactionId = transaction.getId();
+          if (transactionIdFuture != null && !transactionIdFuture.isDone()) {
+            transactionIdFuture.set(transactionId);
+          }
+        }
+      } finally {
+        txnLock.unlock();
+      }
+    }
+
+    @Override
+    public SpannerException onError(
+        SpannerException e, boolean withBeginTransaction, boolean lastStatement) {
+      e = super.onError(e, withBeginTransaction, lastStatement);
+      if (withBeginTransaction) {
+        failTransactionIdFuture(e);
+      }
+      return e;
+    }
+
+    @Override
+    public void onDone(boolean withBeginTransaction) {
+      if (withBeginTransaction) {
+        failTransactionIdFuture(
+            SpannerExceptionFactory.newSpannerException(
+                ErrorCode.FAILED_PRECONDITION,
+                "ResultSet was closed before a read-only transaction id was returned"));
+      }
+      super.onDone(withBeginTransaction);
+    }
+
+    @Override
+    void onStartFailed(boolean withBeginTransaction, Throwable t) {
+      if (withBeginTransaction) {
+        failTransactionIdFuture(t);
+      }
+    }
+
+    private void failTransactionIdFuture(Throwable t) {
+      txnLock.lock();
+      try {
+        if (transactionIdFuture != null && !transactionIdFuture.isDone()) {
+          transactionIdFuture.setException(t);
+        }
+      } finally {
+        txnLock.unlock();
+      }
+    }
+
     @Override
     public Timestamp getReadTimestamp() {
       txnLock.lock();
@@ -544,6 +682,19 @@ public void close() {
       super.close();
     }
 
+    private TransactionOptions createReadOnlyTransactionOptions() {
+      TransactionOptions.Builder options = TransactionOptions.newBuilder();
+      if (timestamp != null) {
+        options
+            .getReadOnlyBuilder()
+            .setReadTimestamp(timestamp.toProto())
+            .setReturnReadTimestamp(true);
+      } else {
+        bound.applyToBuilder(options.getReadOnlyBuilder()).setReturnReadTimestamp(true);
+      }
+      return options.build();
+    }
+
     /**
      * Initializes the transaction with the timestamp specified within MultiUseReadOnlyTransaction.
      * This is used only for fallback of PartitionQueryRequest and PartitionReadRequest with
@@ -553,19 +704,10 @@ void initFallbackTransaction() {
       txnLock.lock();
       try {
         span.addAnnotation("Creating Transaction");
-        TransactionOptions.Builder options = TransactionOptions.newBuilder();
-        if (timestamp != null) {
-          options
-              .getReadOnlyBuilder()
-              .setReadTimestamp(timestamp.toProto())
-              .setReturnReadTimestamp(true);
-        } else {
-          bound.applyToBuilder(options.getReadOnlyBuilder()).setReturnReadTimestamp(true);
-        }
         final BeginTransactionRequest request =
             BeginTransactionRequest.newBuilder()
                 .setSession(session.getName())
-                .setOptions(options)
+                .setOptions(createReadOnlyTransactionOptions())
                 .build();
         initTransactionInternal(request);
       } finally {
@@ -589,12 +731,10 @@ void initTransaction() {
           return;
         }
         span.addAnnotation("Creating Transaction");
-        TransactionOptions.Builder options = TransactionOptions.newBuilder();
-        bound.applyToBuilder(options.getReadOnlyBuilder()).setReturnReadTimestamp(true);
         final BeginTransactionRequest request =
             BeginTransactionRequest.newBuilder()
                 .setSession(session.getName())
-                .setOptions(options)
+                .setOptions(createReadOnlyTransactionOptions())
                 .build();
         initTransactionInternal(request);
       } finally {
@@ -992,15 +1132,22 @@ CloseableIterator<PartialResultSet> startStream(
             if (selector != null) {
               request.setTransaction(selector);
             }
-            SpannerRpc.StreamingCall call =
-                rpc.executeQuery(
-                    request.build(),
-                    stream.consumer(),
-                    getTransactionChannelHint(),
-                    requestId,
-                    isRouteToLeader());
+            boolean withBeginTransaction = request.getTransaction().hasBegin();
+            SpannerRpc.StreamingCall call;
+            try {
+              call =
+                  rpc.executeQuery(
+                      request.build(),
+                      stream.consumer(),
+                      getTransactionChannelHint(),
+                      requestId,
+                      isRouteToLeader());
+            } catch (RuntimeException | Error t) {
+              onStartFailed(withBeginTransaction, t);
+              throw t;
+            }
             session.markUsed(clock.instant());
-            stream.setCall(call, request.getTransaction().hasBegin());
+            stream.setCall(call, withBeginTransaction);
             return stream;
           }
 
@@ -1116,6 +1263,8 @@ public void onDone(boolean withBeginTransaction) {
     this.session.onReadDone();
   }
 
+  void onStartFailed(boolean withBeginTransaction, Throwable t) {}
+
   /**
    * For transactions other than read-write, the MultiplexedSessionPrecommitToken will not be
    * present in the RPC response. In such cases, this method will be a no-op.
@@ -1215,15 +1364,22 @@ CloseableIterator<PartialResultSet> startStream(
               builder.setTransaction(selector);
             }
             builder.setRequestOptions(buildRequestOptions(readOptions));
-            SpannerRpc.StreamingCall call =
-                rpc.read(
-                    builder.build(),
-                    stream.consumer(),
-                    getTransactionChannelHint(),
-                    requestId,
-                    isRouteToLeader());
+            boolean withBeginTransaction = builder.getTransaction().hasBegin();
+            SpannerRpc.StreamingCall call;
+            try {
+              call =
+                  rpc.read(
+                      builder.build(),
+                      stream.consumer(),
+                      getTransactionChannelHint(),
+                      requestId,
+                      isRouteToLeader());
+            } catch (RuntimeException | Error t) {
+              onStartFailed(withBeginTransaction, t);
+              throw t;
+            }
             session.markUsed(clock.instant());
-            stream.setCall(call, /* withBeginTransaction= */ builder.getTransaction().hasBegin());
+            stream.setCall(call, withBeginTransaction);
             return stream;
           }
 

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BuiltInMetricsConstant.java

@@ -144,6 +144,7 @@ public class BuiltInMetricsConstant {
       ImmutableList.of(EEF_FALLBACK_COUNT_NAME, EEF_CALL_STATUS_NAME);
 
   public static final String SPANNER_RESOURCE_TYPE = "spanner_instance_client";
+  public static final String UNDEFINED_PROJECT_ID = "undefined-project";
 
   public static final AttributeKey<String> PROJECT_ID_KEY = AttributeKey.stringKey("project_id");
   public static final AttributeKey<String> INSTANCE_ID_KEY = AttributeKey.stringKey("instance_id");