diff --git a/java-spanner/google-cloud-spanner/clirr-ignored-differences.xml b/java-spanner/google-cloud-spanner/clirr-ignored-differences.xml index 6878a46db1d7..cdcb343a6a01 100644 --- a/java-spanner/google-cloud-spanner/clirr-ignored-differences.xml +++ b/java-spanner/google-cloud-spanner/clirr-ignored-differences.xml @@ -1139,4 +1139,21 @@ com/google/cloud/spanner/XGoogSpannerRequestId REQUEST_ID + + + 7002 + com/google/cloud/spanner/spi/v1/ChannelFinder + com.google.cloud.spanner.spi.v1.ChannelEndpoint fillRoutingHint(com.google.spanner.v1.CommitRequest$Builder) + + + 7002 + com/google/cloud/spanner/spi/v1/ChannelFinder + com.google.cloud.spanner.spi.v1.ChannelEndpoint fillRoutingHint(com.google.spanner.v1.CommitRequest$Builder, java.util.function.Predicate) + + + 7002 + com/google/cloud/spanner/spi/v1/ChannelFinder + * findServer(*) + diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/ChannelFinder.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/ChannelFinder.java index f4fdb41fe3c5..0d7b43e2f820 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/ChannelFinder.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/ChannelFinder.java @@ -250,23 +250,26 @@ void awaitPendingUpdates() throws InterruptedException { } } - public ChannelEndpoint findServer(ReadRequest.Builder reqBuilder) { - return findServer(reqBuilder, preferLeader(reqBuilder.getTransaction()), NO_EXCLUDED_ENDPOINTS); + public ChannelEndpoint findServerAndFillRoutingHint(ReadRequest.Builder reqBuilder) { + return findServerAndFillRoutingHint( + reqBuilder, preferLeader(reqBuilder.getTransaction()), NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint findServer( + public ChannelEndpoint findServerAndFillRoutingHint( ReadRequest.Builder reqBuilder, Predicate excludedEndpoints) { - return findServer(reqBuilder, preferLeader(reqBuilder.getTransaction()), excludedEndpoints); + return findServerAndFillRoutingHint( + reqBuilder, preferLeader(reqBuilder.getTransaction()), excludedEndpoints); } - public ChannelEndpoint findServer(ReadRequest.Builder reqBuilder, boolean preferLeader) { - return findServer(reqBuilder, preferLeader, NO_EXCLUDED_ENDPOINTS); + public ChannelEndpoint findServerAndFillRoutingHint( + ReadRequest.Builder reqBuilder, boolean preferLeader) { + return findServerAndFillRoutingHint(reqBuilder, preferLeader, NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint findServer( + public ChannelEndpoint findServerAndFillRoutingHint( ReadRequest.Builder reqBuilder, boolean preferLeader, Predicate excludedEndpoints) { recipeCache.computeKeys(reqBuilder); - return fillRoutingHint( + return fillRoutingHintAndSelectEndpoint( preferLeader, KeyRangeCache.RangeMode.COVERING_SPLIT, reqBuilder.getDirectedReadOptions(), @@ -274,25 +277,28 @@ public ChannelEndpoint findServer( excludedEndpoints); } - public ChannelEndpoint findServer(ExecuteSqlRequest.Builder reqBuilder) { - return findServer(reqBuilder, preferLeader(reqBuilder.getTransaction()), NO_EXCLUDED_ENDPOINTS); + public ChannelEndpoint findServerAndFillRoutingHint(ExecuteSqlRequest.Builder reqBuilder) { + return findServerAndFillRoutingHint( + reqBuilder, preferLeader(reqBuilder.getTransaction()), NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint findServer( + public ChannelEndpoint findServerAndFillRoutingHint( ExecuteSqlRequest.Builder reqBuilder, Predicate excludedEndpoints) { - return findServer(reqBuilder, preferLeader(reqBuilder.getTransaction()), excludedEndpoints); + return findServerAndFillRoutingHint( + reqBuilder, preferLeader(reqBuilder.getTransaction()), excludedEndpoints); } - public ChannelEndpoint findServer(ExecuteSqlRequest.Builder reqBuilder, boolean preferLeader) { - return findServer(reqBuilder, preferLeader, NO_EXCLUDED_ENDPOINTS); + public ChannelEndpoint findServerAndFillRoutingHint( + ExecuteSqlRequest.Builder reqBuilder, boolean preferLeader) { + return findServerAndFillRoutingHint(reqBuilder, preferLeader, NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint findServer( + public ChannelEndpoint findServerAndFillRoutingHint( ExecuteSqlRequest.Builder reqBuilder, boolean preferLeader, Predicate excludedEndpoints) { recipeCache.computeKeys(reqBuilder); - return fillRoutingHint( + return fillRoutingHintAndSelectEndpoint( preferLeader, KeyRangeCache.RangeMode.PICK_RANDOM, reqBuilder.getDirectedReadOptions(), @@ -300,11 +306,33 @@ public ChannelEndpoint findServer( excludedEndpoints); } - public ChannelEndpoint findServer(BeginTransactionRequest.Builder reqBuilder) { - return findServer(reqBuilder, NO_EXCLUDED_ENDPOINTS); + /** + * Fills the routing hint of {@code reqBuilder} (operation UID, schema generation, database ID, + * key, target range and tablet) without selecting an endpoint for the request. Use this when the + * request is already bound to an endpoint, for example by read-write transaction affinity. + */ + public void fillRoutingHint( + ReadRequest.Builder reqBuilder, boolean preferLeader, Predicate excludedEndpoints) { + findServerAndFillRoutingHint(reqBuilder, preferLeader, excludedEndpoints); + } + + /** + * Fills the routing hint of {@code reqBuilder} (operation UID, schema generation, database ID, + * key, target range and tablet) without selecting an endpoint for the request. Use this when the + * request is already bound to an endpoint, for example by read-write transaction affinity. + */ + public void fillRoutingHint( + ExecuteSqlRequest.Builder reqBuilder, + boolean preferLeader, + Predicate excludedEndpoints) { + findServerAndFillRoutingHint(reqBuilder, preferLeader, excludedEndpoints); + } + + public ChannelEndpoint findServerAndFillRoutingHint(BeginTransactionRequest.Builder reqBuilder) { + return findServerAndFillRoutingHint(reqBuilder, NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint findServer( + public ChannelEndpoint findServerAndFillRoutingHint( BeginTransactionRequest.Builder reqBuilder, Predicate excludedEndpoints) { if (!reqBuilder.hasMutationKey()) { return null; @@ -316,11 +344,11 @@ public ChannelEndpoint findServer( excludedEndpoints); } - public ChannelEndpoint fillRoutingHint(CommitRequest.Builder reqBuilder) { - return fillRoutingHint(reqBuilder, NO_EXCLUDED_ENDPOINTS); + public ChannelEndpoint findServerAndFillRoutingHint(CommitRequest.Builder reqBuilder) { + return findServerAndFillRoutingHint(reqBuilder, NO_EXCLUDED_ENDPOINTS); } - public ChannelEndpoint fillRoutingHint( + public ChannelEndpoint findServerAndFillRoutingHint( CommitRequest.Builder reqBuilder, Predicate excludedEndpoints) { Mutation mutation = selectMutationForRouting(reqBuilder.getMutationsList()); if (mutation == null) { @@ -365,7 +393,7 @@ private ChannelEndpoint routeMutation( return null; } recipeCache.applyTargetRange(hintBuilder, target); - return fillRoutingHint( + return fillRoutingHintAndSelectEndpoint( preferLeader, KeyRangeCache.RangeMode.COVERING_SPLIT, DirectedReadOptions.getDefaultInstance(), @@ -373,7 +401,7 @@ private ChannelEndpoint routeMutation( excludedEndpoints); } - private ChannelEndpoint fillRoutingHint( + private ChannelEndpoint fillRoutingHintAndSelectEndpoint( boolean preferLeader, KeyRangeCache.RangeMode rangeMode, DirectedReadOptions directedReadOptions, diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java index b109ea321a8c..0344188c2ff0 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java @@ -831,7 +831,7 @@ public void sendMessage(RequestT message) { databaseScope = databaseId; } if (finder != null && reqBuilder.hasMutationKey()) { - endpoint = finder.findServer(reqBuilder, excludedEndpoints); + endpoint = finder.findServerAndFillRoutingHint(reqBuilder, excludedEndpoints); } if (reqBuilder.hasOptions() && reqBuilder.getOptions().hasReadOnly()) { shouldRecordTransactionAffinity = false; @@ -850,7 +850,7 @@ public void sendMessage(RequestT message) { CommitRequest.Builder reqBuilder = null; if (finder != null && request.getMutationsCount() > 0) { reqBuilder = request.toBuilder(); - endpoint = finder.fillRoutingHint(reqBuilder, excludedEndpoints); + endpoint = finder.findServerAndFillRoutingHint(reqBuilder, excludedEndpoints); request = reqBuilder.build(); } if (!request.getTransactionId().isEmpty()) { @@ -1162,8 +1162,13 @@ private RoutingDecision routeFromRequest(ReadRequest.Builder reqBuilder) { // by this request's routing hint without leader bias, as read-only transaction IDs can run // on any suitable replica. boolean routePreferLeader = transactionId.isEmpty() && preferLeader; - endpoint = finder.findServer(reqBuilder, routePreferLeader, excludedEndpoints); + endpoint = + finder.findServerAndFillRoutingHint(reqBuilder, routePreferLeader, excludedEndpoints); preferLeader = routePreferLeader; + } else if (finder != null) { + // Pinned to the read-write transaction's endpoint, but still send a routing hint so the + // server can use the fastpath and return key recipes for this statement. + finder.fillRoutingHint(reqBuilder, preferLeader, excludedEndpoints); } return new RoutingDecision( finder, endpoint, databaseId, operationUid(reqBuilder.getRoutingHint()), preferLeader); @@ -1184,8 +1189,13 @@ private RoutingDecision routeFromRequest(ExecuteSqlRequest.Builder reqBuilder) { // by this request's routing hint without leader bias, as read-only transaction IDs can run // on any suitable replica. boolean routePreferLeader = transactionId.isEmpty() && preferLeader; - endpoint = finder.findServer(reqBuilder, routePreferLeader, excludedEndpoints); + endpoint = + finder.findServerAndFillRoutingHint(reqBuilder, routePreferLeader, excludedEndpoints); preferLeader = routePreferLeader; + } else if (finder != null) { + // Pinned to the read-write transaction's endpoint, but still send a routing hint so the + // server can use the fastpath and return key recipes for this statement. + finder.fillRoutingHint(reqBuilder, preferLeader, excludedEndpoints); } return new RoutingDecision( finder, endpoint, databaseId, operationUid(reqBuilder.getRoutingHint()), preferLeader); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/ChannelFinderGoldenTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/ChannelFinderGoldenTest.java index a5b5338b0e06..df3457a19464 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/ChannelFinderGoldenTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/ChannelFinderGoldenTest.java @@ -79,13 +79,13 @@ public void goldenTest() throws Exception { switch (event.getRequestCase()) { case READ: ReadRequest.Builder readBuilder = event.getRead().toBuilder(); - ChannelEndpoint readEndpoint = finder.findServer(readBuilder); + ChannelEndpoint readEndpoint = finder.findServerAndFillRoutingHint(readBuilder); assertHintAndServer( testCase.getName(), event, readBuilder.getRoutingHint(), readEndpoint); break; case SQL: ExecuteSqlRequest.Builder sqlBuilder = event.getSql().toBuilder(); - ChannelEndpoint sqlEndpoint = finder.findServer(sqlBuilder); + ChannelEndpoint sqlEndpoint = finder.findServerAndFillRoutingHint(sqlBuilder); assertHintAndServer( testCase.getName(), event, sqlBuilder.getRoutingHint(), sqlEndpoint); break; diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java index b4293fe64b2a..21d6ac10dd9a 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/KeyAwareChannelTest.java @@ -1314,6 +1314,78 @@ public void unaryReadWithTransactionIdRoutesToAffinityEndpoint() throws Exceptio assertThat(harness.defaultManagedChannel.callCount()).isEqualTo(1); } + @Test + public void readWriteTransactionReadAttachesRoutingHintWhilePinnedToAffinityEndpoint() + throws Exception { + TestHarness harness = createHarness(); + ByteString transactionId = ByteString.copyFromUtf8("rw-read-routing-hint"); + seedReadWriteTransactionAffinity(harness, transactionId); + + // Key "n" is owned by server-b, but the transaction is pinned to server-a. + ClientCall call = + harness.channel.newCall(SpannerGrpc.getStreamingReadMethod(), CallOptions.DEFAULT); + call.start(new CapturingListener(), new Metadata()); + call.sendMessage( + ReadRequest.newBuilder() + .setSession(SESSION) + .setTable("T") + .setTransaction(TransactionSelector.newBuilder().setId(transactionId)) + .setRoutingHint(RoutingHint.newBuilder().setKey(bytes("n"))) + .build()); + + assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(2); + assertThat(harness.endpointCache.callCountForAddress("server-b:1234")).isEqualTo(0); + @SuppressWarnings("unchecked") + RecordingClientCall delegate = + (RecordingClientCall) + harness.endpointCache.latestCallForAddress("server-a:1234"); + RoutingHint routingHint = delegate.lastMessage.getRoutingHint(); + assertThat(routingHint.getOperationUid()).isNotEqualTo(0L); + assertThat(routingHint.getDatabaseId()).isEqualTo(7L); + // The hint describes the split covering key "n" ([m, z), group 2) and its leader tablet. + assertThat(routingHint.getKey()).isEqualTo(bytes("m")); + assertThat(routingHint.getLimitKey()).isEqualTo(bytes("z")); + assertThat(routingHint.getGroupUid()).isEqualTo(2L); + assertThat(routingHint.getSplitId()).isEqualTo(2L); + assertThat(routingHint.getTabletUid()).isEqualTo(2L); + } + + @Test + public void readWriteTransactionQueryAttachesRoutingHintWhilePinnedToAffinityEndpoint() + throws Exception { + TestHarness harness = createHarness(); + ByteString transactionId = ByteString.copyFromUtf8("rw-query-routing-hint"); + seedReadWriteTransactionAffinity(harness, transactionId); + + // Key "n" is owned by server-b, but the transaction is pinned to server-a. + ClientCall call = + harness.channel.newCall(SpannerGrpc.getExecuteStreamingSqlMethod(), CallOptions.DEFAULT); + call.start(new CapturingListener(), new Metadata()); + call.sendMessage( + ExecuteSqlRequest.newBuilder() + .setSession(SESSION) + .setSql("SELECT * FROM T WHERE K = @k") + .setTransaction(TransactionSelector.newBuilder().setId(transactionId)) + .setRoutingHint(RoutingHint.newBuilder().setKey(bytes("n"))) + .build()); + + assertThat(harness.endpointCache.callCountForAddress("server-a:1234")).isEqualTo(2); + assertThat(harness.endpointCache.callCountForAddress("server-b:1234")).isEqualTo(0); + @SuppressWarnings("unchecked") + RecordingClientCall delegate = + (RecordingClientCall) + harness.endpointCache.latestCallForAddress("server-a:1234"); + RoutingHint routingHint = delegate.lastMessage.getRoutingHint(); + assertThat(routingHint.getOperationUid()).isNotEqualTo(0L); + assertThat(routingHint.getDatabaseId()).isEqualTo(7L); + // The hint describes the split covering key "n" ([m, z), group 2) and its leader tablet. + assertThat(routingHint.getKey()).isEqualTo(bytes("m")); + assertThat(routingHint.getLimitKey()).isEqualTo(bytes("z")); + assertThat(routingHint.getGroupUid()).isEqualTo(2L); + assertThat(routingHint.getSplitId()).isEqualTo(2L); + assertThat(routingHint.getTabletUid()).isEqualTo(2L); + } + @Test public void genericUnimplementedReroutesOnceAndLearnsDefaultOnlyMethod() throws Exception { TestHarness harness = createHarness();