@@ -535,8 +535,6 @@ private ResultSet executeQueryInternal(
 * <li>Specific {@link QueryOptions} passed in for this query.
 * <li>Any value specified in a valid environment variable when the {@link SpannerOptions}
 * instance was created.
-* <li>The default {@link SpannerOptions#getDefaultQueryOptions()} specified for the database
-* where the query is executed.
 * </ol>
 */
 @VisibleForTesting
@@ -554,7 +552,8 @@ QueryOptions buildQueryOptions(QueryOptions requestOptions) {
 return builder.build();
 }
 
-ExecuteSqlRequest.Builder getExecuteSqlRequestBuilder(Statement statement, QueryMode queryMode) {
+ExecuteSqlRequest.Builder getExecuteSqlRequestBuilder(
+Statement statement, QueryMode queryMode, boolean withTransactionSelector) {
 ExecuteSqlRequest.Builder builder =
 ExecuteSqlRequest.newBuilder()
 .setSql(statement.getSql())
@@ -568,9 +567,11 @@ ExecuteSqlRequest.Builder getExecuteSqlRequestBuilder(Statement statement, Query
 builder.putParamTypes(param.getKey(), param.getValue().getType().toProto());
 }
 }
-TransactionSelector selector = getTransactionSelector();
-if (selector != null) {
-builder.setTransaction(selector);
+if (withTransactionSelector) {
+TransactionSelector selector = getTransactionSelector();
+if (selector != null) {
+builder.setTransaction(selector);
+}
 }
 builder.setSeqno(getSeqNo());
 builder.setQueryOptions(buildQueryOptions(statement.getQueryOptions()));
@@ -614,18 +615,26 @@ ResultSet executeQueryInternalWithOptions(
 beforeReadOrQuery();
 final int prefetchChunks =
 options.hasPrefetchChunks() ? options.prefetchChunks() : defaultPrefetchChunks;
+final ExecuteSqlRequest.Builder request =
+getExecuteSqlRequestBuilder(
+statement, queryMode, /* withTransactionSelector = */ false);
 ResumableStreamIterator stream =
 new ResumableStreamIterator(MAX_BUFFERED_CHUNKS, SpannerImpl.QUERY, span) {
 @Override
 CloseableIterator<PartialResultSet> startStream(@Nullable ByteString resumeToken) {
 GrpcStreamIterator stream = new GrpcStreamIterator(statement, prefetchChunks);
-final ExecuteSqlRequest.Builder request =
-getExecuteSqlRequestBuilder(statement, queryMode);
 if (partitionToken != null) {
 request.setPartitionToken(partitionToken);
 }
+TransactionSelector selector = null;
 if (resumeToken != null) {
 request.setResumeToken(resumeToken);
+selector = getTransactionSelector();
+} else if (!request.hasTransaction()) {
+selector = getTransactionSelector();
+}
+if (selector != null) {
+request.setTransaction(selector);
 }
 SpannerRpc.Call call =
 rpc.executeQuery(request.build(), stream.consumer(), session.getOptions());
@@ -733,10 +742,13 @@ ResultSet readInternalWithOptions(
 @Override
 CloseableIterator<PartialResultSet> startStream(@Nullable ByteString resumeToken) {
 GrpcStreamIterator stream = new GrpcStreamIterator(prefetchChunks);
+TransactionSelector selector = null;
 if (resumeToken != null) {
 builder.setResumeToken(resumeToken);
+selector = getTransactionSelector();
+} else if (!builder.hasTransaction()) {
+selector = getTransactionSelector();
 }
-TransactionSelector selector = getTransactionSelector();
 if (selector != null) {
 builder.setTransaction(selector);
 }

@@ -1071,6 +1071,7 @@ protected PartialResultSet computeNext() {
 backoffSleep(context, backOff);
 }
 }
+
 continue;
 }
 span.addAnnotation("Stream broken. Not safe to retry");

@@ -515,7 +515,10 @@ public void buffer(Iterable<Mutation> mutations) {
 public long executeUpdate(Statement statement) {
 beforeReadOrQuery();
 final ExecuteSqlRequest.Builder builder =
-getExecuteSqlRequestBuilder(statement, QueryMode.NORMAL);
+getExecuteSqlRequestBuilder(
+statement,
+QueryMode.NORMAL,
+/* withTransactionSelector = */ true);
 try {
 com.google.spanner.v1.ResultSet resultSet =
 rpc.executeQuery(builder.build(), session.getOptions());
@@ -538,7 +541,10 @@ public long executeUpdate(Statement statement) {
 public ApiFuture<Long> executeUpdateAsync(Statement statement) {
 beforeReadOrQuery();
 final ExecuteSqlRequest.Builder builder =
-getExecuteSqlRequestBuilder(statement, QueryMode.NORMAL);
+getExecuteSqlRequestBuilder(
+statement,
+QueryMode.NORMAL,
+/* withTransactionSelector = */ true);
 final ApiFuture<com.google.spanner.v1.ResultSet> resultSet;
 try {
 // Register the update as an async operation that must finish before the transaction may

@@ -89,7 +89,10 @@ public void setup() {
 public void executeSqlRequestBuilderWithoutQueryOptions() {
 ExecuteSqlRequest request =
 context
-.getExecuteSqlRequestBuilder(Statement.of("SELECT FOO FROM BAR"), QueryMode.NORMAL)
+.getExecuteSqlRequestBuilder(
+Statement.of("SELECT FOO FROM BAR"),
+QueryMode.NORMAL,
+true)
 .build();
 assertThat(request.getSql()).isEqualTo("SELECT FOO FROM BAR");
 assertThat(request.getQueryOptions()).isEqualTo(defaultQueryOptions);
@@ -103,7 +106,8 @@ public void executeSqlRequestBuilderWithQueryOptions() {
 Statement.newBuilder("SELECT FOO FROM BAR")
 .withQueryOptions(QueryOptions.newBuilder().setOptimizerVersion("2.0").build())
 .build(),
-QueryMode.NORMAL)
+QueryMode.NORMAL,
+true)
 .build();
 assertThat(request.getSql()).isEqualTo("SELECT FOO FROM BAR");
 assertThat(request.getQueryOptions().getOptimizerVersion()).isEqualTo("2.0");

@@ -251,6 +251,130 @@ public Long run(TransactionContext transaction) throws Exception {
 assertThat(countTransactionsStarted()).isEqualTo(2);
 }
 
+@Test
+public void testInlinedBeginFirstUpdateAborts() {
+DatabaseClient client = spanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
+long updateCount =
+client
+.readWriteTransaction()
+.run(
+new TransactionCallable<Long>() {
+boolean firstAttempt = true;
+
+@Override
+public Long run(TransactionContext transaction) throws Exception {
+if (firstAttempt) {
+firstAttempt = false;
+mockSpanner.putStatementResult(
+StatementResult.exception(
+UPDATE_STATEMENT,
+mockSpanner.createAbortedException(
+ByteString.copyFromUtf8("some-tx"))));
+} else {
+mockSpanner.putStatementResult(
+StatementResult.update(UPDATE_STATEMENT, UPDATE_COUNT));
+}
+return transaction.executeUpdate(UPDATE_STATEMENT);
+}
+});
+assertThat(updateCount).isEqualTo(UPDATE_COUNT);
+assertThat(countRequests(BeginTransactionRequest.class)).isEqualTo(1);
+assertThat(countRequests(ExecuteSqlRequest.class)).isEqualTo(2);
+assertThat(countRequests(CommitRequest.class)).isEqualTo(1);
+}
+
+@Test
+public void testInlinedBeginFirstQueryAborts() {
+DatabaseClient client = spanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
+long updateCount =
+client
+.readWriteTransaction()
+.run(
+new TransactionCallable<Long>() {
+boolean firstAttempt = true;
+
+@Override
+public Long run(TransactionContext transaction) throws Exception {
+if (firstAttempt) {
+firstAttempt = false;
+mockSpanner.putStatementResult(
+StatementResult.exception(
+SELECT1,
+mockSpanner.createAbortedException(
+ByteString.copyFromUtf8("some-tx"))));
+} else {
+mockSpanner.putStatementResult(
+StatementResult.query(SELECT1, SELECT1_RESULTSET));
+}
+try (ResultSet rs = transaction.executeQuery(SELECT1)) {
+while (rs.next()) {
+return rs.getLong(0);
+}
+}
+return 0L;
+}
+});
+assertThat(updateCount).isEqualTo(1L);
+assertThat(countRequests(BeginTransactionRequest.class)).isEqualTo(1);
+assertThat(countRequests(ExecuteSqlRequest.class)).isEqualTo(2);
+assertThat(countRequests(CommitRequest.class)).isEqualTo(1);
+}
+
+@Test
+public void testInlinedBeginFirstQueryReturnsUnavailable() {
+DatabaseClient client = spanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
+mockSpanner.setExecuteSqlExecutionTime(
+SimulatedExecutionTime.ofStreamException(Status.UNAVAILABLE.asRuntimeException(), 0));
+long value =
+client
+.readWriteTransaction()
+.run(
+new TransactionCallable<Long>() {
+@Override
+public Long run(TransactionContext transaction) throws Exception {
+// The first attempt will return UNAVAILABLE and retry internally.
+try (ResultSet rs = transaction.executeQuery(SELECT1)) {
+while (rs.next()) {
+return rs.getLong(0);
+}
+}
+return 0L;
+}
+});
+assertThat(value).isEqualTo(1L);
+assertThat(countRequests(BeginTransactionRequest.class)).isEqualTo(0);
+assertThat(countRequests(ExecuteSqlRequest.class)).isEqualTo(2);
+assertThat(countRequests(CommitRequest.class)).isEqualTo(1);
+}
+
+@Test
+public void testInlinedBeginFirstReadReturnsUnavailable() {
+DatabaseClient client = spanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
+mockSpanner.setReadExecutionTime(
+SimulatedExecutionTime.ofStreamException(Status.UNAVAILABLE.asRuntimeException(), 0));
+long value =
+client
+.readWriteTransaction()
+.run(
+new TransactionCallable<Long>() {
+@Override
+public Long run(TransactionContext transaction) throws Exception {
+// The first attempt will return UNAVAILABLE and retry internally.
+try (ResultSet rs =
+transaction.read("FOO", KeySet.all(), Arrays.asList("ID"))) {
+while (rs.next()) {
+return rs.getLong(0);
+}
+}
+return 0L;
+}
+});
+assertThat(value).isEqualTo(1L);
+assertThat(countRequests(BeginTransactionRequest.class)).isEqualTo(0);
+assertThat(countRequests(ReadRequest.class)).isEqualTo(2);
+assertThat(countRequests(CommitRequest.class)).isEqualTo(1);
+}
+
 @Test
 public void testInlinedBeginTxWithQuery() {
 DatabaseClient client =
@@ -279,8 +403,7 @@ public Long run(TransactionContext transaction) throws Exception {
 
 @Test
 public void testInlinedBeginTxWithRead() {
-DatabaseClient client =
-spanner.getDatabaseClient(DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]"));
+DatabaseClient client = spanner.getDatabaseClient(DatabaseId.of("p", "i", "d"));
 long updateCount =
 client
 .readWriteTransaction()