Skip to content
This repository was archived by the owner on May 8, 2026. It is now read-only.

Commit 84b1f20

Browse files
mutianfmutianf
committed
fix: fix skip large row
1 parent de1669e commit 84b1f20

3 files changed

Lines changed: 263 additions & 48 deletions

File tree

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

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ public static RowSet eraseLargeRow(RowSet rowSet, ByteString excludePoint) {
6868
// Handle ranges
6969
for (RowRange rowRange : rowSet.getRowRangesList()) {
7070
List<RowRange> afterSplit = splitOnLargeRowKey(rowRange, excludePoint);
71-
if (afterSplit != null && !afterSplit.isEmpty()) {
71+
if (!afterSplit.isEmpty()) {
7272
afterSplit.forEach(newRowSet::addRowRanges);
7373
}
7474
}
@@ -162,8 +162,10 @@ private static List<RowRange> splitOnLargeRowKey(RowRange range, ByteString larg
162162
ByteString startKey = StartPoint.extract(range).value;
163163
ByteString endKey = EndPoint.extract(range).value;
164164

165-
// if end key is on the left of large row key, don't split
166-
if (ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) < 0) {
165+
// if end key is on the left of large row key, don't split. Empty endKey means it's unbounded
166+
// and it's always
167+
// on the right of the large key
168+
if (!endKey.isEmpty() && ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) < 0) {
167169
rowRanges.add(range);
168170
return rowRanges;
169171
}
@@ -181,8 +183,8 @@ private static List<RowRange> splitOnLargeRowKey(RowRange range, ByteString larg
181183
}
182184

183185
// if the end key is on the right of the large row key, set the start key to be large row key
184-
// open
185-
if (ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) > 0) {
186+
// open. Empty end key is unbounded so it's always on the right of the large key
187+
if (endKey.isEmpty() || ByteStringComparator.INSTANCE.compare(endKey, largeRowKey) > 0) {
186188
RowRange afterSplit = range.toBuilder().setStartKeyOpen(largeRowKey).build();
187189
rowRanges.add(afterSplit);
188190
}

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

Lines changed: 130 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -114,51 +114,13 @@ public void testWriteRead() throws Exception {
114114
assertThat(row.getCells().get(1).getValue()).isEqualTo(largeValue);
115115
}
116116

117-
static class AccumulatingObserver implements ResponseObserver<Row> {
118-
119-
final List<Row> responses = Lists.newArrayList();
120-
final SettableApiFuture<Void> completionFuture = SettableApiFuture.create();
121-
122-
void awaitCompletion() throws Throwable {
123-
try {
124-
completionFuture.get(10, TimeUnit.MINUTES);
125-
} catch (ExecutionException e) {
126-
throw e.getCause();
127-
}
128-
}
129-
130-
@Override
131-
public void onStart(StreamController controller) {}
132-
133-
@Override
134-
public void onResponse(Row row) {
135-
responses.add(row);
136-
}
137-
138-
@Override
139-
public void onError(Throwable t) {
140-
completionFuture.setException(t);
141-
}
142-
143-
@Override
144-
public void onComplete() {
145-
completionFuture.set(null);
146-
}
147-
}
148-
149117
@Test
150-
public void read() throws Throwable {
118+
public void testSkipLargeRow() throws Throwable {
151119
assume()
152120
.withMessage("Large row read errors are not supported by emulator")
153121
.that(testEnvRule.env())
154122
.isNotInstanceOf(EmulatorEnv.class);
155123

156-
// TODO: remove this once skip large row for read is released
157-
assume()
158-
.withMessage("Skip large row for read is not released yet")
159-
.that(System.getProperty("bigtable.testSkipLargeRowIntegrationTests"))
160-
.isEqualTo("true");
161-
162124
BigtableDataClient client = testEnvRule.env().getDataClient();
163125
String tableId = table.getId();
164126
String familyId = this.familyId;
@@ -202,12 +164,34 @@ public void read() throws Throwable {
202164
ImmutableList.<String>of(),
203165
ByteString.copyFromUtf8("my-value"))));
204166

167+
Row expectedRow5 =
168+
Row.create(
169+
ByteString.copyFromUtf8("r5"),
170+
ImmutableList.of(
171+
RowCell.create(
172+
familyId,
173+
ByteString.copyFromUtf8("qualifier"),
174+
timestampMicros,
175+
ImmutableList.of(),
176+
ByteString.copyFromUtf8("my-value"))));
177+
178+
Row expectedRow6 =
179+
Row.create(
180+
ByteString.copyFromUtf8("r6"),
181+
ImmutableList.of(
182+
RowCell.create(
183+
familyId,
184+
ByteString.copyFromUtf8("qualifier"),
185+
timestampMicros,
186+
ImmutableList.of(),
187+
ByteString.copyFromUtf8("my-value"))));
188+
205189
// large row creation
206190
byte[] largeValueBytes = new byte[3 * 1024 * 1024];
207191
ByteString largeValue = ByteString.copyFrom(largeValueBytes);
208192

209193
for (int i = 0; i < 100; i++) {
210-
ByteString qualifier = ByteString.copyFromUtf8("qualifier1_" + "_" + String.valueOf(i));
194+
ByteString qualifier = ByteString.copyFromUtf8("qualifier1_" + "_" + i);
211195
client.mutateRow(
212196
RowMutation.create(TableId.of(tableId), "r2").setCell(familyId, qualifier, largeValue));
213197
client.mutateRow(
@@ -222,7 +206,8 @@ public void read() throws Throwable {
222206
.call(
223207
Query.create(tableId)
224208
.range(ByteStringRange.unbounded().startClosed("r1").endOpen("r3"))))
225-
.containsExactly(expectedRow1);
209+
.containsExactly(expectedRow1)
210+
.inOrder();
226211

227212
assertThat(
228213
client
@@ -231,7 +216,8 @@ public void read() throws Throwable {
231216
.call(
232217
Query.create(tableId)
233218
.range(ByteStringRange.unbounded().startClosed("r1").endClosed("r4"))))
234-
.containsExactly(expectedRow1, expectedRow4);
219+
.containsExactly(expectedRow1, expectedRow4)
220+
.inOrder();
235221

236222
List<Row> emptyRows =
237223
client
@@ -267,7 +253,69 @@ public void read() throws Throwable {
267253
.call(
268254
Query.create(tableId)
269255
.range(ByteStringRange.unbounded().startClosed("r1").endClosed("r4"))))
270-
.containsExactly(expectedRow1, expectedRow4);
256+
.containsExactly(expectedRow1, expectedRow4)
257+
.inOrder();
258+
259+
assertThat(client.skipLargeRowsCallable().all().call(Query.create(tableId)))
260+
.containsExactly(expectedRow1, expectedRow4, expectedRow5, expectedRow6)
261+
.inOrder();
262+
263+
assertThat(
264+
client
265+
.skipLargeRowsCallable()
266+
.all()
267+
.call(Query.create(tableId).range(ByteStringRange.unbounded().endClosed("r4"))))
268+
.containsExactly(expectedRow1, expectedRow4)
269+
.inOrder();
270+
271+
assertThat(
272+
client
273+
.skipLargeRowsCallable()
274+
.all()
275+
.call(Query.create(tableId).range(ByteStringRange.unbounded().startClosed("r1"))))
276+
.containsExactly(expectedRow1, expectedRow4, expectedRow5, expectedRow6)
277+
.inOrder();
278+
279+
assertThat(
280+
client
281+
.skipLargeRowsCallable()
282+
.all()
283+
.call(Query.create(tableId).range(ByteStringRange.unbounded().endOpen("r4"))))
284+
.containsExactly(expectedRow1);
285+
286+
assertThat(
287+
client
288+
.skipLargeRowsCallable()
289+
.all()
290+
.call(Query.create(tableId).range(ByteStringRange.unbounded().startOpen("r1"))))
291+
.containsExactly(expectedRow4, expectedRow5, expectedRow6);
292+
293+
assertThat(client.skipLargeRowsCallable().all().call(Query.create(tableId).reversed(true)))
294+
.containsExactly(expectedRow6, expectedRow5, expectedRow4, expectedRow1)
295+
.inOrder();
296+
297+
assertThat(
298+
client
299+
.skipLargeRowsCallable()
300+
.all()
301+
.call(
302+
Query.create(tableId)
303+
.range(ByteStringRange.unbounded().endClosed("r4"))
304+
.reversed(true)))
305+
.containsExactly(expectedRow4, expectedRow1)
306+
.inOrder();
307+
308+
assertThat(
309+
client
310+
.skipLargeRowsCallable()
311+
.all()
312+
.call(
313+
Query.create(tableId)
314+
.range(ByteStringRange.unbounded().startClosed("r1"))
315+
.reversed(true)))
316+
.containsExactly(expectedRow6, expectedRow5, expectedRow4, expectedRow1)
317+
.inOrder();
318+
271319
// async
272320
AccumulatingObserver observer = new AccumulatingObserver();
273321
Query query = Query.create(tableId).range("r1", "r3");
@@ -280,5 +328,44 @@ public void read() throws Throwable {
280328
client.skipLargeRowsCallable().call(query2, observer2);
281329
observer2.awaitCompletion();
282330
assertThat(observer2.responses).containsExactly(expectedRow1, expectedRow4);
331+
332+
AccumulatingObserver observer3 = new AccumulatingObserver();
333+
Query query3 = Query.create(tableId);
334+
client.skipLargeRowsCallable().call(query3, observer3);
335+
observer3.awaitCompletion();
336+
assertThat(observer3.responses)
337+
.containsExactly(expectedRow1, expectedRow4, expectedRow5, expectedRow6);
338+
}
339+
340+
static class AccumulatingObserver implements ResponseObserver<Row> {
341+
342+
final List<Row> responses = Lists.newArrayList();
343+
final SettableApiFuture<Void> completionFuture = SettableApiFuture.create();
344+
345+
void awaitCompletion() throws Throwable {
346+
try {
347+
completionFuture.get(10, TimeUnit.MINUTES);
348+
} catch (ExecutionException e) {
349+
throw e.getCause();
350+
}
351+
}
352+
353+
@Override
354+
public void onStart(StreamController controller) {}
355+
356+
@Override
357+
public void onResponse(Row row) {
358+
responses.add(row);
359+
}
360+
361+
@Override
362+
public void onError(Throwable t) {
363+
completionFuture.setException(t);
364+
}
365+
366+
@Override
367+
public void onComplete() {
368+
completionFuture.set(null);
369+
}
283370
}
284371
}

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

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -335,6 +335,132 @@ public void readRowRangeWithSkippingLargeRows() {
335335
Truth.assertThat(actualResults).containsExactly("r6", "r5", "r4").inOrder();
336336
}
337337

338+
@Test
339+
public void readRowRangeWithUnboundedRanges() {
340+
ApiException largeRowException = createLargeRowException("r3");
341+
342+
// Test case 1: Full table scan (unbounded start and end)
343+
service.expectations.add(
344+
RpcExpectation.create()
345+
.respondWith("r1", "r2")
346+
.respondWithException(Code.INTERNAL, largeRowException));
347+
348+
// After the large row error, the query should be split into two ranges
349+
service.expectations.add(
350+
RpcExpectation.create()
351+
.expectRequestForMultipleRowRanges(
352+
ImmutableList.of(Range.open("r2", "r3"), Range.greaterThan("r3")))
353+
.respondWith("r4")
354+
.respondWithStatus(Code.OK));
355+
356+
List<String> actualResults = getSkipLargeRowsResults(Query.create(TABLE_ID));
357+
Truth.assertThat(actualResults).containsExactly("r1", "r2", "r4").inOrder();
358+
359+
// Test case 2: Unbounded end
360+
service.expectations.add(
361+
RpcExpectation.create()
362+
.expectRequest(Range.atLeast("r2"))
363+
.respondWith("r2")
364+
.respondWithException(Code.INTERNAL, largeRowException));
365+
service.expectations.add(
366+
RpcExpectation.create()
367+
.expectRequestForMultipleRowRanges(
368+
ImmutableList.of(Range.open("r2", "r3"), Range.greaterThan("r3")))
369+
.respondWith("r4")
370+
.respondWithStatus(Code.OK));
371+
372+
actualResults =
373+
getSkipLargeRowsResults(
374+
Query.create(TABLE_ID).range(ByteStringRange.unbounded().startClosed("r2")));
375+
Truth.assertThat(actualResults).containsExactly("r2", "r4").inOrder();
376+
377+
// Test case 3: Unbounded start
378+
service.expectations.add(
379+
RpcExpectation.create()
380+
.expectRequest(Range.atMost("r4"))
381+
.respondWith("r1", "r2")
382+
.respondWithException(Code.INTERNAL, largeRowException));
383+
384+
service.expectations.add(
385+
RpcExpectation.create()
386+
.expectRequestForMultipleRowRanges(
387+
ImmutableList.of(Range.open("r2", "r3"), Range.openClosed("r3", "r4")))
388+
.respondWith("r4")
389+
.respondWithStatus(Code.OK));
390+
391+
actualResults =
392+
getSkipLargeRowsResults(
393+
Query.create(TABLE_ID).range(ByteStringRange.unbounded().endClosed("r4")));
394+
Truth.assertThat(actualResults).containsExactly("r1", "r2", "r4").inOrder();
395+
}
396+
397+
@Test
398+
public void readRowRangeWithUnboundedRangesReversed() {
399+
ApiException largeRowException = createLargeRowException("r3");
400+
401+
// Test case 1: Full table scan (unbounded start and end) reversed
402+
service.expectations.add(
403+
RpcExpectation.create()
404+
.setReversed(true)
405+
.respondWith("r5", "r4")
406+
.respondWithException(Code.INTERNAL, largeRowException));
407+
408+
// After the large row error, the query should be split into two ranges and retried
409+
service.expectations.add(
410+
RpcExpectation.create()
411+
.expectRequestForMultipleRowRanges(
412+
ImmutableList.of(Range.lessThan("r3"), Range.open("r3", "r4")))
413+
.setReversed(true)
414+
.respondWith("r2", "r1")
415+
.respondWithStatus(Code.OK));
416+
417+
List<String> actualResults = getSkipLargeRowsResults(Query.create(TABLE_ID).reversed(true));
418+
Truth.assertThat(actualResults).containsExactly("r5", "r4", "r2", "r1").inOrder();
419+
420+
// Test case 2: Unbounded start reversed
421+
service.expectations.add(
422+
RpcExpectation.create()
423+
.expectRequest(Range.atLeast("r2"))
424+
.setReversed(true)
425+
.respondWith("r5", "r4")
426+
.respondWithException(Code.INTERNAL, largeRowException));
427+
service.expectations.add(
428+
RpcExpectation.create()
429+
.expectRequestForMultipleRowRanges(
430+
ImmutableList.of(Range.closedOpen("r2", "r3"), Range.open("r3", "r4")))
431+
.setReversed(true)
432+
.respondWith("r2")
433+
.respondWithStatus(Code.OK));
434+
actualResults =
435+
getSkipLargeRowsResults(
436+
Query.create(TABLE_ID)
437+
.range(ByteStringRange.unbounded().startClosed("r2"))
438+
.reversed(true));
439+
Truth.assertThat(actualResults).containsExactly("r5", "r4", "r2").inOrder();
440+
441+
// Test case 3: Unbounded end reversed
442+
service.expectations.add(
443+
RpcExpectation.create()
444+
.expectRequest(Range.atMost("r4"))
445+
.setReversed(true)
446+
.respondWith("r4")
447+
.respondWithException(Code.INTERNAL, largeRowException));
448+
449+
service.expectations.add(
450+
RpcExpectation.create()
451+
.expectRequestForMultipleRowRanges(
452+
ImmutableList.of(Range.lessThan("r3"), Range.open("r3", "r4")))
453+
.setReversed(true)
454+
.respondWith("r2", "r1")
455+
.respondWithStatus(Code.OK));
456+
actualResults =
457+
getSkipLargeRowsResults(
458+
Query.create(TABLE_ID)
459+
.range(ByteStringRange.unbounded().endClosed("r4"))
460+
.reversed(true));
461+
Truth.assertThat(actualResults).containsExactly("r4", "r2", "r1").inOrder();
462+
}
463+
338464
@Test
339465
public void multipleRetryTest() {
340466
service.expectations.add(

0 commit comments

Comments
 (0)