Visitar URL original
feat: reverse scans public preview (#1711) · benjaminp/google-cloud-java@bb36292 · GitHub
Skip to content

Commit bb36292

Browse files
igorbernstein2igorbernstein2
authored
feat: reverse scans public preview (googleapis#1711)
This adds a reversed boolean to Query, which will allow endusers to stream rows in reverse order. Example: ```java Query query = Query.create("alphabet").range("a", "z").limit(3); ServerStream<Row> results = client.readRows(query); for (Row row : results) { System.out.println(row.getKey().toStringUtf8()); } // Prints z, y, x ```
1 parent cbfa581 commit bb36292

17 files changed

Lines changed: 337 additions & 86 deletions

File tree

‎java-bigtable/google-cloud-bigtable/clirr-ignored-differences.xml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,4 +134,15 @@
134134
<method>*</method>
135135
<to>*</to>
136136
</difference>
137+
<!-- Removed methods in an internal class -->
138+
<difference>
139+
<differenceType>7002</differenceType>
140+
<className>com/google/cloud/bigtable/data/v2/internal/RowSetUtil</className>
141+
<method>*</method>
142+
</difference>
143+
<difference>
144+
<differenceType>7004</differenceType>
145+
<className>com/google/cloud/bigtable/data/v2/stub/readrows/RowMerger</className>
146+
<method>*</method>
147+
</difference>
137148
</differences>

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/RowMergerUtil.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ public class RowMergerUtil implements AutoCloseable {
3030

3131
public RowMergerUtil() {
3232
RowBuilder<Row> rowBuilder = new DefaultRowAdapter().createRowBuilder();
33-
merger = new RowMerger<>(rowBuilder);
33+
merger = new RowMerger<>(rowBuilder, false);
3434
}
3535

3636
@Override

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/RowSetUtil.java‎

Lines changed: 57 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -50,80 +50,79 @@ public final class RowSetUtil {
5050
private RowSetUtil() {}
5151

5252
/**
53-
* Splits the provided {@link RowSet} along the provided splitPoint into 2 segments. The right
54-
* segment will contain all keys that are strictly greater than the splitPoint and all {@link
55-
* RowRange}s truncated to start right after the splitPoint. The primary usecase is to resume a
56-
* broken ReadRows stream.
53+
* Removes all the keys and range parts that fall on or before the splitPoint.
54+
*
55+
* <p>The direction of before is determined by fromStart: for forward scans fromStart is true and
56+
* will remove all the keys and range segments that would've been read prior to the splitPoint
57+
* (ie. all of the keys sort lexiographically at or before the split point. For reverse scans,
58+
* fromStart is false and all segments that sort lexiographically at or after the split point are
59+
* removed. The primary usecase is to resume a broken ReadRows stream.
5760
*/
58-
@Nonnull
59-
public static Split split(@Nonnull RowSet rowSet, @Nonnull ByteString splitPoint) {
60-
// Edgecase: splitPoint is the leftmost key ("")
61-
if (splitPoint.isEmpty()) {
62-
return Split.of(null, rowSet);
63-
}
61+
public static RowSet erase(RowSet rowSet, ByteString splitPoint, boolean fromStart) {
62+
RowSet.Builder newRowSet = RowSet.newBuilder();
6463

65-
// An empty RowSet represents a full table scan. Make that explicit so that there is RowRange to
66-
// split.
6764
if (rowSet.getRowKeysList().isEmpty() && rowSet.getRowRangesList().isEmpty()) {
6865
rowSet = RowSet.newBuilder().addRowRanges(RowRange.getDefaultInstance()).build();
6966
}
7067

71-
RowSet.Builder leftBuilder = RowSet.newBuilder();
72-
boolean leftIsEmpty = true;
73-
RowSet.Builder rightBuilder = RowSet.newBuilder();
74-
boolean rightIsEmpty = true;
75-
68+
// Handle point lookups
7669
for (ByteString key : rowSet.getRowKeysList()) {
77-
if (ByteStringComparator.INSTANCE.compare(key, splitPoint) <= 0) {
78-
leftBuilder.addRowKeys(key);
79-
leftIsEmpty = false;
70+
if (fromStart) {
71+
// key is right of the split
72+
if (ByteStringComparator.INSTANCE.compare(key, splitPoint) > 0) {
73+
newRowSet.addRowKeys(key);
74+
}
8075
} else {
81-
rightBuilder.addRowKeys(key);
82-
rightIsEmpty = false;
76+
// key is left of the split
77+
if (ByteStringComparator.INSTANCE.compare(key, splitPoint) < 0) {
78+
newRowSet.addRowKeys(key);
79+
}
8380
}
8481
}
8582

86-
for (RowRange range : rowSet.getRowRangesList()) {
87-
StartPoint startPoint = StartPoint.extract(range);
88-
int startCmp =
89-
ComparisonChain.start()
90-
.compare(startPoint.value, splitPoint, ByteStringComparator.INSTANCE)
91-
// when value lies on the split point, only closed start points are on the left
92-
.compareTrueFirst(startPoint.isClosed, true)
93-
.result();
94-
95-
// Range is fully on the right side
96-
if (startCmp > 0) {
97-
rightBuilder.addRowRanges(range);
98-
rightIsEmpty = false;
99-
continue;
83+
// Handle ranges
84+
for (RowRange rowRange : rowSet.getRowRangesList()) {
85+
RowRange newRange = truncateRange(rowRange, splitPoint, fromStart);
86+
if (newRange != null) {
87+
newRowSet.addRowRanges(newRange);
10088
}
89+
}
10190

102-
EndPoint endPoint = EndPoint.extract(range);
103-
int endCmp =
104-
ComparisonChain.start()
105-
// empty (true) end key means rightmost regardless of the split point
106-
.compareFalseFirst(endPoint.value.isEmpty(), false)
107-
.compare(endPoint.value, splitPoint, ByteStringComparator.INSTANCE)
108-
// don't care if the endpoint is open/closed: both will be on the left if the value is
109-
// <=
110-
.result();
111-
112-
if (endCmp <= 0) {
113-
// Range is fully on the left
114-
leftBuilder.addRowRanges(range);
115-
leftIsEmpty = false;
116-
} else {
117-
// Range is split
118-
leftBuilder.addRowRanges(range.toBuilder().setEndKeyClosed(splitPoint));
119-
leftIsEmpty = false;
120-
rightBuilder.addRowRanges(range.toBuilder().setStartKeyOpen(splitPoint));
121-
rightIsEmpty = false;
91+
// Return the new rowset if there is anything left to read
92+
RowSet result = newRowSet.build();
93+
if (result.getRowKeysList().isEmpty() && result.getRowRangesList().isEmpty()) {
94+
return null;
95+
}
96+
return result;
97+
}
98+
99+
private static RowRange truncateRange(RowRange range, ByteString split, boolean fromStart) {
100+
if (fromStart) {
101+
// range end is on or left of the split: skip
102+
if (EndPoint.extract(range).compareTo(new EndPoint(split, true)) <= 0) {
103+
return null;
104+
}
105+
} else {
106+
// range is on or right of the split
107+
if (StartPoint.extract(range).compareTo(new StartPoint(split, true)) >= 0) {
108+
return null;
109+
}
110+
}
111+
RowRange.Builder newRange = range.toBuilder();
112+
113+
if (fromStart) {
114+
// range start is on or left of the split
115+
if (StartPoint.extract(range).compareTo(new StartPoint(split, true)) <= 0) {
116+
newRange.setStartKeyOpen(split);
117+
}
118+
} else {
119+
// range end is on or right of the split
120+
if (EndPoint.extract(range).compareTo(new EndPoint(split, true)) >= 0) {
121+
newRange.setEndKeyOpen(split);
122122
}
123123
}
124124

125-
return Split.of(
126-
leftIsEmpty ? null : leftBuilder.build(), rightIsEmpty ? null : rightBuilder.build());
125+
return newRange.build();
127126
}
128127

129128
/**

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Query.java‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,26 @@ public Query limit(long limit) {
184184
return this;
185185
}
186186

187+
/**
188+
* Return rows in reverse order.
189+
*
190+
* <p>The row will be streamed in reverse lexiographic order of the keys. The row key ranges are
191+
* still expected to be oriented the same way as forwards. ie [a,c] where a <= c. The row content
192+
* will remain unchanged from the ordering forward scans. This is particularly useful to get the
193+
* last N records before a key:
194+
*
195+
* <pre>{@code
196+
* query
197+
* .range(ByteStringRange.unbounded().endOpen("key"))
198+
* .limit(10)
199+
* .reversed(true)
200+
* }</pre>
201+
*/
202+
public Query reversed(boolean enable) {
203+
builder.setReversed(enable);
204+
return this;
205+
}
206+
187207
/**
188208
* Split this query into multiple queries that can be evenly distributed across Bigtable nodes and
189209
* be run in parallel. This method takes the results from {@link
@@ -379,11 +399,12 @@ public boolean advance(@Nonnull ByteString lastSeenRowKey) {
379399

380400
// Split the row ranges / row keys. Return false if there's nothing
381401
// left on the right of the split point.
382-
RowSetUtil.Split split = RowSetUtil.split(query.builder.getRows(), lastSeenRowKey);
383-
if (split.getRight() == null) {
402+
RowSet remaining =
403+
RowSetUtil.erase(query.builder.getRows(), lastSeenRowKey, !query.builder.getReversed());
404+
if (remaining == null) {
384405
return false;
385406
}
386-
query.builder.setRows(split.getRight());
407+
query.builder.setRows(remaining);
387408
return true;
388409
}
389410
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -732,7 +732,7 @@ private Builder() {
732732
.setTotalTimeout(PRIME_REQUEST_TIMEOUT)
733733
.build());
734734

735-
featureFlags = FeatureFlags.newBuilder();
735+
featureFlags = FeatureFlags.newBuilder().setReverseScans(true);
736736
}
737737

738738
private Builder(EnhancedBigtableStubSettings settings) {

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,8 @@ public ReadRowsRequest getResumeRequest(ReadRowsRequest originalRequest) {
8585
return originalRequest;
8686
}
8787

88-
RowSet remaining = RowSetUtil.split(originalRequest.getRows(), lastKey).getRight();
88+
RowSet remaining =
89+
RowSetUtil.erase(originalRequest.getRows(), lastKey, !originalRequest.getReversed());
8990

9091
// Edge case: retrying a fulfilled request.
9192
// A fulfilled request is one that has had all of its row keys and ranges fulfilled, or if it

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -61,8 +61,8 @@ public class RowMerger<RowT> implements Reframer<RowT, ReadRowsResponse> {
6161
private final StateMachine<RowT> stateMachine;
6262
private Queue<RowT> mergedRows;
6363

64-
public RowMerger(RowBuilder<RowT> rowBuilder) {
65-
stateMachine = new StateMachine<>(rowBuilder);
64+
public RowMerger(RowBuilder<RowT> rowBuilder, boolean reversed) {
65+
stateMachine = new StateMachine<>(rowBuilder, reversed);
6666
mergedRows = new ArrayDeque<>();
6767
}
6868

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public RowMergingCallable(
4949
public void call(
5050
ReadRowsRequest request, ResponseObserver<RowT> responseObserver, ApiCallContext context) {
5151
RowBuilder<RowT> rowBuilder = rowAdapter.createRowBuilder();
52-
RowMerger<RowT> merger = new RowMerger<>(rowBuilder);
52+
RowMerger<RowT> merger = new RowMerger<>(rowBuilder, request.getReversed());
5353
ReframingResponseObserver<ReadRowsResponse, RowT> innerObserver =
5454
new ReframingResponseObserver<>(responseObserver, merger);
5555
inner.call(request, innerObserver, context);

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@
7676
*/
7777
final class StateMachine<RowT> {
7878
private final RowBuilder<RowT> adapter;
79+
private boolean reversed;
7980
private State currentState;
8081
private ByteString lastCompleteRowKey;
8182

@@ -102,9 +103,11 @@ final class StateMachine<RowT> {
102103
* Initialize a new state machine that's ready for a new row.
103104
*
104105
* @param adapter The adapter that will build the final row.
106+
* @param reversed
105107
*/
106-
StateMachine(RowBuilder<RowT> adapter) {
108+
StateMachine(RowBuilder<RowT> adapter, boolean reversed) {
107109
this.adapter = adapter;
110+
this.reversed = reversed;
108111
reset();
109112
}
110113

@@ -261,9 +264,15 @@ State handleChunk(CellChunk chunk) {
261264
validate(chunk.hasFamilyName(), "AWAITING_NEW_ROW: family missing");
262265
validate(chunk.hasQualifier(), "AWAITING_NEW_ROW: qualifier missing");
263266
if (lastCompleteRowKey != null) {
264-
validate(
265-
ByteStringComparator.INSTANCE.compare(lastCompleteRowKey, chunk.getRowKey()) < 0,
266-
"AWAITING_NEW_ROW: key must be strictly increasing");
267+
268+
int cmp = ByteStringComparator.INSTANCE.compare(lastCompleteRowKey, chunk.getRowKey());
269+
String direction = "increasing";
270+
if (reversed) {
271+
cmp *= -1;
272+
direction = "decreasing";
273+
}
274+
275+
validate(cmp < 0, "AWAITING_NEW_ROW: key must be strictly " + direction);
267276
}
268277

269278
rowKey = chunk.getRowKey();

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

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.google.api.gax.rpc.TransportChannelProvider;
2727
import com.google.api.gax.rpc.WatchdogProvider;
2828
import com.google.bigtable.v2.BigtableGrpc;
29+
import com.google.bigtable.v2.FeatureFlags;
2930
import com.google.bigtable.v2.InstanceName;
3031
import com.google.bigtable.v2.MutateRowRequest;
3132
import com.google.bigtable.v2.MutateRowResponse;
@@ -36,8 +37,14 @@
3637
import com.google.cloud.bigtable.data.v2.internal.NameUtil;
3738
import com.google.cloud.bigtable.data.v2.models.RowMutation;
3839
import com.google.common.base.Preconditions;
40+
import com.google.common.io.BaseEncoding;
3941
import io.grpc.Attributes;
42+
import io.grpc.Metadata;
4043
import io.grpc.Server;
44+
import io.grpc.ServerCall;
45+
import io.grpc.ServerCall.Listener;
46+
import io.grpc.ServerCallHandler;
47+
import io.grpc.ServerInterceptor;
4148
import io.grpc.ServerTransportFilter;
4249
import io.grpc.stub.StreamObserver;
4350
import java.io.IOException;
@@ -78,12 +85,24 @@ public class BigtableDataClientFactoryTest {
7885

7986
private final BlockingQueue<Attributes> setUpAttributes = new LinkedBlockingDeque<>();
8087
private final BlockingQueue<Attributes> terminateAttributes = new LinkedBlockingDeque<>();
88+
private final BlockingQueue<Metadata> requestMetadata = new LinkedBlockingDeque<>();
8189

8290
@Before
8391
public void setUp() throws IOException {
8492
service = new FakeBigtableService();
8593
server =
8694
FakeServiceBuilder.create(service)
95+
.intercept(
96+
new ServerInterceptor() {
97+
@Override
98+
public <ReqT, RespT> Listener<ReqT> interceptCall(
99+
ServerCall<ReqT, RespT> call,
100+
Metadata headers,
101+
ServerCallHandler<ReqT, RespT> next) {
102+
requestMetadata.add(headers);
103+
return next.startCall(call, headers);
104+
}
105+
})
87106
.addTransportFilter(
88107
new ServerTransportFilter() {
89108
@Override
@@ -276,6 +295,24 @@ public void testCreateWithRefreshingChannel() throws Exception {
276295
assertThat(terminateAttributes).hasSize(poolSize);
277296
}
278297

298+
@Test
299+
public void testFeatureFlags() throws Exception {
300+
try (BigtableDataClientFactory factory = BigtableDataClientFactory.create(defaultSettings);
301+
BigtableDataClient client = factory.createDefault()) {
302+
303+
requestMetadata.clear();
304+
client.mutateRow(RowMutation.create("some-table", "some-key").deleteRow());
305+
}
306+
307+
Metadata metadata = requestMetadata.take();
308+
String encodedValue =
309+
metadata.get(Metadata.Key.of("bigtable-features", Metadata.ASCII_STRING_MARSHALLER));
310+
FeatureFlags featureFlags =
311+
FeatureFlags.parseFrom(BaseEncoding.base64Url().decode(encodedValue));
312+
313+
assertThat(featureFlags.getReverseScans()).isTrue();
314+
}
315+
279316
@Test
280317
public void testBulkMutationFlowControllerConfigured() throws Exception {
281318
BigtableDataSettings settings =
@@ -306,6 +343,7 @@ private static class FakeBigtableService extends BigtableGrpc.BigtableImplBase {
306343
volatile MutateRowRequest lastRequest;
307344
BlockingQueue<ReadRowsRequest> readRowsRequests = new LinkedBlockingDeque<>();
308345
BlockingQueue<PingAndWarmRequest> pingAndWarmRequests = new LinkedBlockingDeque<>();
346+
309347
private ApiFunction<ReadRowsRequest, ReadRowsResponse> readRowsCallback =
310348
new ApiFunction<ReadRowsRequest, ReadRowsResponse>() {
311349
@Override

0 commit comments

Comments
 (0)