Visitar URL original
feat: expose new API with ReadRowsRequest in EnhancedBigtableStub (#276) · googleapis/java-bigtable@394efe4 · GitHub
Skip to content
This repository was archived by the owner on Aug 30, 2026. It is now read-only.

Commit 394efe4

Browse files
rahulKQLrahulKQL
authored
feat: expose new API with ReadRowsRequest in EnhancedBigtableStub (#276)
* feat: expose StreamingCallable with ReadRowsRequest in EnhancedBigtableStub This commit would enable the user to target the table using absolute resource name on each read request. Currently we expose `ServerStreamingCallable<Query, RowT>`, which does not have an option to provide different `app-profile-id` on each request. * chore: reorganized createReadRows callable methods With this commit, the public createReadRowsCallable() would refer to single createReadRowsBaseCallable.
1 parent 4374492 commit 394efe4

2 files changed

Lines changed: 190 additions & 21 deletions

File tree

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java‎

Lines changed: 44 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package com.google.cloud.bigtable.data.v2.stub;
1717

18+
import com.google.api.core.BetaApi;
1819
import com.google.api.core.InternalApi;
1920
import com.google.api.gax.batching.Batcher;
2021
import com.google.api.gax.batching.BatcherImpl;
@@ -148,6 +149,27 @@ private EnhancedBigtableStub(
148149

149150
// <editor-fold desc="Callable creators">
150151

152+
/**
153+
* Creates a callable chain to handle ReadRows RPCs. The chain will:
154+
*
155+
* <ul>
156+
* <li>Dispatch the RPC with {@link ReadRowsRequest}.
157+
* <li>Upon receiving the response stream, it will merge the {@link
158+
* com.google.bigtable.v2.ReadRowsResponse.CellChunk}s in logical rows. The actual row
159+
* implementation can be configured by the {@code rowAdapter} parameter.
160+
* <li>Retry/resume on failure.
161+
* <li>Filter out marker rows.
162+
* </ul>
163+
*
164+
* <p>NOTE: the caller is responsible for adding tracing & metrics.
165+
*/
166+
@BetaApi("This surface is stable yet it might be removed in the future.")
167+
public <RowT> ServerStreamingCallable<ReadRowsRequest, RowT> createReadRowsRawCallable(
168+
RowAdapter<RowT> rowAdapter) {
169+
return createReadRowsBaseCallable(settings.readRowsSettings(), rowAdapter)
170+
.withDefaultCallContext(clientContext.getDefaultCallContext());
171+
}
172+
151173
/**
152174
* Creates a callable chain to handle streaming ReadRows RPCs. The chain will:
153175
*
@@ -164,12 +186,15 @@ private EnhancedBigtableStub(
164186
*/
165187
public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
166188
RowAdapter<RowT> rowAdapter) {
167-
ServerStreamingCallable<Query, RowT> readRowsCallable =
189+
ServerStreamingCallable<ReadRowsRequest, RowT> readRowsCallable =
168190
createReadRowsBaseCallable(settings.readRowsSettings(), rowAdapter);
169191

192+
ServerStreamingCallable<Query, RowT> readRowsUserCallable =
193+
new ReadRowsUserCallable<>(readRowsCallable, requestContext);
194+
170195
ServerStreamingCallable<Query, RowT> traced =
171196
new TracedServerStreamingCallable<>(
172-
readRowsCallable,
197+
readRowsUserCallable,
173198
clientContext.getTracerFactory(),
174199
SpanName.of(TRACING_OUTER_CLIENT_NAME, "ReadRows"));
175200

@@ -199,15 +224,17 @@ public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
199224
* </ul>
200225
*/
201226
public <RowT> UnaryCallable<Query, RowT> createReadRowCallable(RowAdapter<RowT> rowAdapter) {
202-
UnaryCallable<Query, RowT> readRowCallable =
227+
ServerStreamingCallable<ReadRowsRequest, RowT> readRowsCallable =
203228
createReadRowsBaseCallable(
204-
ServerStreamingCallSettings.<Query, Row>newBuilder()
205-
.setRetryableCodes(settings.readRowSettings().getRetryableCodes())
206-
.setRetrySettings(settings.readRowSettings().getRetrySettings())
207-
.setIdleTimeout(settings.readRowSettings().getRetrySettings().getTotalTimeout())
208-
.build(),
209-
rowAdapter)
210-
.first();
229+
ServerStreamingCallSettings.<ReadRowsRequest, Row>newBuilder()
230+
.setRetryableCodes(settings.readRowSettings().getRetryableCodes())
231+
.setRetrySettings(settings.readRowSettings().getRetrySettings())
232+
.setIdleTimeout(settings.readRowSettings().getRetrySettings().getTotalTimeout())
233+
.build(),
234+
rowAdapter);
235+
236+
UnaryCallable<Query, RowT> readRowCallable =
237+
new ReadRowsUserCallable<>(readRowsCallable, requestContext).first();
211238

212239
return createUserFacingUnaryCallable("ReadRow", readRowCallable);
213240
}
@@ -216,19 +243,18 @@ public <RowT> UnaryCallable<Query, RowT> createReadRowCallable(RowAdapter<RowT>
216243
* Creates a callable chain to handle ReadRows RPCs. The chain will:
217244
*
218245
* <ul>
219-
* <li>Convert a {@link Query} into a {@link com.google.bigtable.v2.ReadRowsRequest} and
220-
* dispatch the RPC.
246+
* <li>Dispatch the RPC with {@link ReadRowsRequest}.
221247
* <li>Upon receiving the response stream, it will merge the {@link
222248
* com.google.bigtable.v2.ReadRowsResponse.CellChunk}s in logical rows. The actual row
223-
* implementation can be configured in by the {@code rowAdapter} parameter.
249+
* implementation can be configured by the {@code rowAdapter} parameter.
224250
* <li>Retry/resume on failure.
225251
* <li>Filter out marker rows.
226252
* </ul>
227253
*
228254
* <p>NOTE: the caller is responsible for adding tracing & metrics.
229255
*/
230-
private <RowT> ServerStreamingCallable<Query, RowT> createReadRowsBaseCallable(
231-
ServerStreamingCallSettings<Query, Row> readRowsSettings, RowAdapter<RowT> rowAdapter) {
256+
private <ReqT, RowT> ServerStreamingCallable<ReadRowsRequest, RowT> createReadRowsBaseCallable(
257+
ServerStreamingCallSettings<ReqT, Row> readRowsSettings, RowAdapter<RowT> rowAdapter) {
232258

233259
ServerStreamingCallable<ReadRowsRequest, ReadRowsResponse> base =
234260
GrpcRawCallableFactory.createServerStreamingCallable(
@@ -249,8 +275,8 @@ public Map<String, String> extract(ReadRowsRequest readRowsRequest) {
249275
ServerStreamingCallable<ReadRowsRequest, RowT> merging =
250276
new RowMergingCallable<>(base, rowAdapter);
251277

252-
// Copy settings for the middle ReadRowsRequest -> RowT callable (as opposed to the outer
253-
// Query -> RowT callable or the inner ReadRowsRequest -> ReadRowsResponse callable).
278+
// Copy settings for the middle ReadRowsRequest -> RowT callable (as opposed to the inner
279+
// ReadRowsRequest -> ReadRowsResponse callable).
254280
ServerStreamingCallSettings<ReadRowsRequest, RowT> innerSettings =
255281
ServerStreamingCallSettings.<ReadRowsRequest, RowT>newBuilder()
256282
.setResumptionStrategy(new ReadRowsResumptionStrategy<>(rowAdapter))
@@ -270,10 +296,7 @@ public Map<String, String> extract(ReadRowsRequest readRowsRequest) {
270296
ServerStreamingCallable<ReadRowsRequest, RowT> retrying2 =
271297
Callables.retrying(retrying1, innerSettings, clientContext);
272298

273-
FilterMarkerRowsCallable<RowT> filtering =
274-
new FilterMarkerRowsCallable<>(retrying2, rowAdapter);
275-
276-
return new ReadRowsUserCallable<>(filtering, requestContext);
299+
return new FilterMarkerRowsCallable<>(retrying2, rowAdapter);
277300
}
278301

279302
/**

‎google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubTest.java‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
/*
2+
* Copyright 2020 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.google.cloud.bigtable.data.v2.stub;
17+
18+
import static com.google.common.truth.Truth.assertThat;
19+
20+
import com.google.api.gax.core.NoCredentialsProvider;
21+
import com.google.api.gax.grpc.testing.InProcessServer;
22+
import com.google.api.gax.grpc.testing.LocalChannelProvider;
23+
import com.google.api.gax.rpc.ServerStreamingCallable;
24+
import com.google.bigtable.v2.BigtableGrpc;
25+
import com.google.bigtable.v2.ReadRowsRequest;
26+
import com.google.bigtable.v2.ReadRowsResponse;
27+
import com.google.bigtable.v2.RowSet;
28+
import com.google.cloud.bigtable.admin.v2.internal.NameUtil;
29+
import com.google.cloud.bigtable.data.v2.internal.RequestContext;
30+
import com.google.cloud.bigtable.data.v2.models.DefaultRowAdapter;
31+
import com.google.cloud.bigtable.data.v2.models.Query;
32+
import com.google.cloud.bigtable.data.v2.models.Row;
33+
import com.google.common.collect.Queues;
34+
import com.google.protobuf.ByteString;
35+
import com.google.protobuf.BytesValue;
36+
import com.google.protobuf.StringValue;
37+
import io.grpc.stub.StreamObserver;
38+
import java.io.IOException;
39+
import java.util.concurrent.BlockingQueue;
40+
import java.util.concurrent.TimeUnit;
41+
import org.junit.After;
42+
import org.junit.Before;
43+
import org.junit.Test;
44+
import org.junit.runner.RunWith;
45+
import org.junit.runners.JUnit4;
46+
47+
@RunWith(JUnit4.class)
48+
public class EnhancedBigtableStubTest {
49+
50+
private static final String PROJECT_ID = "fake-project";
51+
private static final String INSTANCE_ID = "fake-instance";
52+
private static final String FAKE_HOST_NAME = "fake-stub-host:123";
53+
private static final String TABLE_NAME =
54+
NameUtil.formatTableName(PROJECT_ID, INSTANCE_ID, "fake-table");
55+
private static final String APP_PROFILE_ID = "app-profile-id";
56+
57+
private InProcessServer<?> server;
58+
private FakeDataService fakeDataService;
59+
private EnhancedBigtableStub enhancedBigtableStub;
60+
61+
@Before
62+
public void setUp() throws IOException, IllegalAccessException, InstantiationException {
63+
fakeDataService = new FakeDataService();
64+
server = new InProcessServer<>(fakeDataService, FAKE_HOST_NAME);
65+
server.start();
66+
67+
EnhancedBigtableStubSettings enhancedBigtableStubSettings =
68+
EnhancedBigtableStubSettings.newBuilder()
69+
.setProjectId(PROJECT_ID)
70+
.setInstanceId(INSTANCE_ID)
71+
.setAppProfileId(APP_PROFILE_ID)
72+
.setCredentialsProvider(NoCredentialsProvider.create())
73+
.setEndpoint(FAKE_HOST_NAME)
74+
.setTransportChannelProvider(LocalChannelProvider.create(FAKE_HOST_NAME))
75+
.build();
76+
77+
enhancedBigtableStub = EnhancedBigtableStub.create(enhancedBigtableStubSettings);
78+
}
79+
80+
@After
81+
public void tearDown() {
82+
server.stop();
83+
}
84+
85+
@Test
86+
public void testCreateReadRowsCallable() throws InterruptedException {
87+
ServerStreamingCallable<Query, Row> streamingCallable =
88+
enhancedBigtableStub.createReadRowsCallable(new DefaultRowAdapter());
89+
90+
Query request = Query.create("table-id").rowKey("row-key");
91+
streamingCallable.call(request).iterator().next();
92+
ReadRowsRequest expected =
93+
request.toProto(RequestContext.create(PROJECT_ID, INSTANCE_ID, APP_PROFILE_ID));
94+
assertThat(fakeDataService.popLastRequest()).isEqualTo(expected);
95+
}
96+
97+
@Test
98+
public void testCreateReadRowsRawCallable() throws InterruptedException {
99+
ServerStreamingCallable<ReadRowsRequest, Row> callable =
100+
enhancedBigtableStub.createReadRowsRawCallable(new DefaultRowAdapter());
101+
102+
ReadRowsRequest expectedRequest =
103+
ReadRowsRequest.newBuilder()
104+
.setTableName(TABLE_NAME)
105+
.setAppProfileId("app-profile-1")
106+
.setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("test-row-key")))
107+
.build();
108+
callable.call(expectedRequest).iterator().next();
109+
assertThat(fakeDataService.popLastRequest()).isEqualTo(expectedRequest);
110+
111+
ReadRowsRequest expectedRequest2 =
112+
ReadRowsRequest.newBuilder()
113+
.setTableName(TABLE_NAME)
114+
.setAppProfileId("app-profile-2")
115+
.build();
116+
callable.call(expectedRequest2).iterator().next();
117+
assertThat(fakeDataService.popLastRequest()).isEqualTo(expectedRequest2);
118+
}
119+
120+
private static class FakeDataService extends BigtableGrpc.BigtableImplBase {
121+
final BlockingQueue<Object> requests = Queues.newLinkedBlockingDeque();
122+
123+
@SuppressWarnings("unchecked")
124+
<T> T popLastRequest() throws InterruptedException {
125+
return (T) requests.poll(1, TimeUnit.SECONDS);
126+
}
127+
128+
@Override
129+
public void readRows(
130+
ReadRowsRequest request, StreamObserver<ReadRowsResponse> responseObserver) {
131+
requests.add(request);
132+
// Dummy row for stream
133+
responseObserver.onNext(
134+
ReadRowsResponse.newBuilder()
135+
.addChunks(
136+
ReadRowsResponse.CellChunk.newBuilder()
137+
.setCommitRow(true)
138+
.setRowKey(ByteString.copyFromUtf8("a"))
139+
.setFamilyName(StringValue.getDefaultInstance())
140+
.setQualifier(BytesValue.getDefaultInstance())
141+
.setValueSize(0))
142+
.build());
143+
responseObserver.onCompleted();
144+
}
145+
}
146+
}

0 commit comments

Comments
 (0)