Skip to content

Commit 5d9c8cb

Browse files
bbeaudreaultbbeaudreault
committed
HBASE-26122: Implement an optional maximum size for Gets, after which a partial result is returned
1 parent 20a4aae commit 5d9c8cb

File tree

9 files changed

+215
-13
lines changed

9 files changed

+215
-13
lines changed

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/Get.java

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ public class Get extends Query implements Row {
7676
private boolean checkExistenceOnly = false;
7777
private boolean closestRowBefore = false;
7878
private Map<byte [], NavigableSet<byte []>> familyMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
79+
private long maxResultSize = -1;
7980

8081
/**
8182
* Create a Get operation for the specified row.
@@ -339,6 +340,21 @@ public Get setFilter(Filter filter) {
339340
return this;
340341
}
341342

343+
/**
344+
* Set the maximum result size. The default is -1; this means that no specific
345+
* maximum result size will be set for this Get.
346+
*
347+
* If set to a value greater than zero, the server may respond with a Result where
348+
* {@link Result#mayHaveMoreCellsInRow()} is true. The user is required to handle
349+
* this case.
350+
*
351+
* @param maxResultSize The maximum result size in bytes
352+
*/
353+
public Get setMaxResultSize(long maxResultSize) {
354+
this.maxResultSize = maxResultSize;
355+
return this;
356+
}
357+
342358
/* Accessors */
343359

344360
/**
@@ -458,6 +474,13 @@ public Map<String, Object> getFingerprint() {
458474
return map;
459475
}
460476

477+
/**
478+
* @return the maximum result size in bytes. See {@link #setMaxResultSize(long)}
479+
*/
480+
public long getMaxResultSize() {
481+
return maxResultSize;
482+
}
483+
461484
/**
462485
* Compile the details beyond the scope of getFingerprint (row, columns,
463486
* timestamps, etc.) into a Map along with the fingerprinted information.

‎hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -434,6 +434,9 @@ public static Get toGet(final ClientProtos.Get proto) throws IOException {
434434
if (proto.hasLoadColumnFamiliesOnDemand()) {
435435
get.setLoadColumnFamiliesOnDemand(proto.getLoadColumnFamiliesOnDemand());
436436
}
437+
if (proto.hasMaxResultSize()) {
438+
get.setMaxResultSize(proto.getMaxResultSize());
439+
}
437440
return get;
438441
}
439442

@@ -1122,6 +1125,10 @@ public static ClientProtos.Get toGet(
11221125
builder.setLoadColumnFamiliesOnDemand(loadColumnFamiliesOnDemand);
11231126
}
11241127

1128+
if (get.getMaxResultSize() > 0) {
1129+
builder.setMaxResultSize(get.getMaxResultSize());
1130+
}
1131+
11251132
return builder.build();
11261133
}
11271134

@@ -1382,7 +1389,7 @@ public static Result toResult(final ClientProtos.Result proto, final CellScanner
13821389

13831390
return (cells == null || cells.isEmpty())
13841391
? (proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT)
1385-
: Result.create(cells, null, proto.getStale());
1392+
: Result.create(cells, null, proto.getStale(), proto.getPartial());
13861393
}
13871394

13881395

‎hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -592,6 +592,9 @@ public static Get toGet(final ClientProtos.Get proto) throws IOException {
592592
if (proto.hasLoadColumnFamiliesOnDemand()) {
593593
get.setLoadColumnFamiliesOnDemand(proto.getLoadColumnFamiliesOnDemand());
594594
}
595+
if (proto.hasMaxResultSize()) {
596+
get.setMaxResultSize(proto.getMaxResultSize());
597+
}
595598
return get;
596599
}
597600

@@ -1256,6 +1259,9 @@ public static ClientProtos.Get toGet(
12561259
if (loadColumnFamiliesOnDemand != null) {
12571260
builder.setLoadColumnFamiliesOnDemand(loadColumnFamiliesOnDemand);
12581261
}
1262+
if (get.getMaxResultSize() > 0) {
1263+
builder.setMaxResultSize(get.getMaxResultSize());
1264+
}
12591265
return builder.build();
12601266
}
12611267

@@ -1457,6 +1463,7 @@ public static ClientProtos.Result toResultNoData(final Result result) {
14571463
ClientProtos.Result.Builder builder = ClientProtos.Result.newBuilder();
14581464
builder.setAssociatedCellCount(size);
14591465
builder.setStale(result.isStale());
1466+
builder.setPartial(result.mayHaveMoreCellsInRow());
14601467
return builder.build();
14611468
}
14621469

@@ -1547,7 +1554,7 @@ public static Result toResult(final ClientProtos.Result proto, final CellScanner
15471554

15481555
return (cells == null || cells.isEmpty())
15491556
? (proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT)
1550-
: Result.create(cells, null, proto.getStale());
1557+
: Result.create(cells, null, proto.getStale(), proto.getPartial());
15511558
}
15521559

15531560

‎hbase-protocol-shaded/src/main/protobuf/Client.proto

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,8 @@ message Get {
9090
optional Consistency consistency = 12 [default = STRONG];
9191
repeated ColumnFamilyTimeRange cf_time_range = 13;
9292
optional bool load_column_families_on_demand = 14; /* DO NOT add defaults to load_column_families_on_demand. */
93+
94+
optional uint64 max_result_size = 15;
9395
}
9496

9597
message Result {

‎hbase-protocol/src/main/protobuf/Client.proto

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,8 @@ message Get {
9191
optional Consistency consistency = 12 [default = STRONG];
9292
repeated ColumnFamilyTimeRange cf_time_range = 13;
9393
optional bool load_column_families_on_demand = 14; /* DO NOT add defaults to load_column_families_on_demand. */
94+
95+
optional uint64 max_result_size = 15;
9496
}
9597

9698
message Result {

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

Lines changed: 31 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@
146146
import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl.WriteEntry;
147147
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
148148
import org.apache.hadoop.hbase.regionserver.compactions.CompactionLifeCycleTracker;
149+
import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
149150
import org.apache.hadoop.hbase.regionserver.throttle.CompactionThroughputControllerFactory;
150151
import org.apache.hadoop.hbase.regionserver.throttle.NoLimitThroughputController;
151152
import org.apache.hadoop.hbase.regionserver.throttle.StoreHotnessProtector;
@@ -3864,8 +3865,7 @@ public void prepareMiniBatchOperations(MiniBatchOperationInProgress<Mutation> mi
38643865
Result result;
38653866
if (returnResults) {
38663867
// convert duplicate increment/append to get
3867-
List<Cell> results = region.get(toGet(mutation), false, nonceGroup, nonce);
3868-
result = Result.create(results);
3868+
result = region.get(toGet(mutation), false, nonceGroup, nonce);
38693869
} else {
38703870
result = Result.EMPTY_RESULT;
38713871
}
@@ -7497,9 +7497,7 @@ public static boolean rowIsInRange(RegionInfo info, final byte [] row, final int
74977497
@Override
74987498
public Result get(final Get get) throws IOException {
74997499
prepareGet(get);
7500-
List<Cell> results = get(get, true);
7501-
boolean stale = this.getRegionInfo().getReplicaId() != 0;
7502-
return Result.create(results, get.isCheckExistenceOnly() ? !results.isEmpty() : null, stale);
7500+
return get(get, true, HConstants.NO_NONCE, HConstants.NO_NONCE);
75037501
}
75047502

75057503
void prepareGet(final Get get) throws IOException {
@@ -7518,11 +7516,35 @@ void prepareGet(final Get get) throws IOException {
75187516

75197517
@Override
75207518
public List<Cell> get(Get get, boolean withCoprocessor) throws IOException {
7521-
return get(get, withCoprocessor, HConstants.NO_NONCE, HConstants.NO_NONCE);
7519+
return getInternal(get, withCoprocessor, HConstants.NO_NONCE, HConstants.NO_NONCE).getFirst();
75227520
}
75237521

7524-
private List<Cell> get(Get get, boolean withCoprocessor, long nonceGroup, long nonce)
7525-
throws IOException {
7522+
private Result get(Get get, boolean withCoprocessor, long nonceGroup, long nonce)
7523+
throws IOException {
7524+
Pair<List<Cell>, ScannerContext> result = getInternal(get, withCoprocessor, nonceGroup, nonce);
7525+
boolean stale = this.getRegionInfo().getReplicaId() != 0;
7526+
7527+
return Result.create(
7528+
result.getFirst(),
7529+
get.isCheckExistenceOnly() ? !result.getFirst().isEmpty() : null,
7530+
stale,
7531+
result.getSecond().mayHaveMoreCellsInRow());
7532+
}
7533+
7534+
private Pair<List<Cell>, ScannerContext> getInternal(Get get, boolean withCoprocessor, long nonceGroup, long nonce)
7535+
throws IOException {
7536+
ScannerContext scannerContext = ScannerContext.newBuilder()
7537+
.setSizeLimit(LimitScope.BETWEEN_CELLS, get.getMaxResultSize(), get.getMaxResultSize())
7538+
.build();
7539+
7540+
return Pair.newPair(
7541+
getInternal(get, scannerContext, withCoprocessor, nonceGroup, nonce),
7542+
scannerContext
7543+
);
7544+
}
7545+
7546+
private List<Cell> getInternal(Get get, ScannerContext scannerContext, boolean withCoprocessor,
7547+
long nonceGroup, long nonce) throws IOException {
75267548
List<Cell> results = new ArrayList<>();
75277549
long before = EnvironmentEdgeManager.currentTime();
75287550

@@ -7539,7 +7561,7 @@ private List<Cell> get(Get get, boolean withCoprocessor, long nonceGroup, long n
75397561
}
75407562
try (RegionScanner scanner = getScanner(scan, null, nonceGroup, nonce)) {
75417563
List<Cell> tmp = new ArrayList<>();
7542-
scanner.next(tmp);
7564+
scanner.next(tmp, scannerContext);
75437565
// Copy EC to heap, then close the scanner.
75447566
// This can be an EXPENSIVE call. It may make an extra copy from offheap to onheap buffers.
75457567
// See more details in HBASE-26036.

‎hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2651,10 +2651,15 @@ private Result get(Get get, HRegion region, RegionScannersCloseCallBack closeCal
26512651
if (scan.getLoadColumnFamiliesOnDemandValue() == null) {
26522652
scan.setLoadColumnFamiliesOnDemand(region.isLoadingCfsOnDemandDefault());
26532653
}
2654+
2655+
ScannerContext scannerContext = ScannerContext.newBuilder()
2656+
.setSizeLimit(LimitScope.BETWEEN_CELLS, get.getMaxResultSize(), get.getMaxResultSize())
2657+
.build();
2658+
26542659
RegionScannerImpl scanner = null;
26552660
try {
26562661
scanner = region.getScanner(scan);
2657-
scanner.next(results);
2662+
scanner.next(results, scannerContext);
26582663
} finally {
26592664
if (scanner != null) {
26602665
if (closeCallBack == null) {
@@ -2679,7 +2684,8 @@ private Result get(Get get, HRegion region, RegionScannersCloseCallBack closeCal
26792684
}
26802685
region.metricsUpdateForGet(results, before);
26812686

2682-
return Result.create(results, get.isCheckExistenceOnly() ? !results.isEmpty() : null, stale);
2687+
return Result.create(results, get.isCheckExistenceOnly() ? !results.isEmpty() : null, stale,
2688+
scannerContext.mayHaveMoreCellsInRow());
26832689
}
26842690

26852691
private void checkBatchSizeAndLogLargeSize(MultiRequest request) throws ServiceException {

‎hbase-server/src/test/java/org/apache/hadoop/hbase/TestPartialResultsFromClientSide.java

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,18 +31,26 @@
3131
import java.util.Set;
3232
import org.apache.hadoop.hbase.client.ClientScanner;
3333
import org.apache.hadoop.hbase.client.Delete;
34+
import org.apache.hadoop.hbase.client.Get;
3435
import org.apache.hadoop.hbase.client.Put;
3536
import org.apache.hadoop.hbase.client.RegionInfo;
3637
import org.apache.hadoop.hbase.client.Result;
3738
import org.apache.hadoop.hbase.client.ResultScanner;
3839
import org.apache.hadoop.hbase.client.Scan;
3940
import org.apache.hadoop.hbase.client.Table;
41+
import org.apache.hadoop.hbase.filter.BinaryComparator;
42+
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
4043
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
4144
import org.apache.hadoop.hbase.filter.ColumnRangeFilter;
45+
import org.apache.hadoop.hbase.filter.FamilyFilter;
4246
import org.apache.hadoop.hbase.filter.Filter;
47+
import org.apache.hadoop.hbase.filter.FilterList;
48+
import org.apache.hadoop.hbase.filter.FilterListWithAND;
4349
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
4450
import org.apache.hadoop.hbase.filter.FirstKeyValueMatchingQualifiersFilter;
51+
import org.apache.hadoop.hbase.filter.PageFilter;
4552
import org.apache.hadoop.hbase.filter.RandomRowFilter;
53+
import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
4654
import org.apache.hadoop.hbase.testclassification.LargeTests;
4755
import org.apache.hadoop.hbase.util.Bytes;
4856
import org.apache.hadoop.hbase.util.ClassSize;
@@ -136,6 +144,46 @@ public static void tearDownAfterClass() throws Exception {
136144
TEST_UTIL.shutdownMiniCluster();
137145
}
138146

147+
@Test
148+
public void testGetPartialResults() throws Exception {
149+
byte[] row = ROWS[0];
150+
151+
Result result;
152+
int cf = 0;
153+
int qf = 0;
154+
int total = 0;
155+
156+
do {
157+
// this will ensure we always return only 1 result
158+
Get get = new Get(row)
159+
.setMaxResultSize(1);
160+
161+
// we want to page through the entire row, this will ensure we always get the next
162+
if (total > 0) {
163+
get.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL,
164+
new ColumnRangeFilter(QUALIFIERS[qf], true, null, false),
165+
new FamilyFilter(CompareOperator.GREATER_OR_EQUAL, new BinaryComparator(FAMILIES[cf]))));
166+
}
167+
168+
// all values are the same, but there should be a value
169+
result = TABLE.get(get);
170+
assertTrue(String.format("Value for family %s (# %s) and qualifier %s (# %s)",
171+
Bytes.toStringBinary(FAMILIES[cf]), cf, Bytes.toStringBinary(QUALIFIERS[qf]), qf),
172+
Bytes.equals(VALUE, result.getValue(FAMILIES[cf], QUALIFIERS[qf])));
173+
174+
total++;
175+
if (++qf >= NUM_QUALIFIERS) {
176+
cf++;
177+
qf = 0;
178+
}
179+
} while (result.mayHaveMoreCellsInRow());
180+
181+
// ensure we iterated all cells in row
182+
assertEquals(NUM_COLS, total);
183+
assertEquals(NUM_FAMILIES, cf);
184+
assertEquals(0, qf);
185+
}
186+
139187
/**
140188
* Ensure that the expected key values appear in a result returned from a scanner that is
141189
* combining partial results into complete results

‎hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7861,4 +7861,89 @@ public void run() {
78617861
assertFalse("Region lock holder should not have been interrupted", holderInterrupted.get());
78627862
}
78637863

7864+
@Test
7865+
public void testOversizedGetsReturnPartialResult() throws IOException {
7866+
HRegion region = initHRegion(tableName, name.getMethodName(), CONF, fam1);
7867+
7868+
Put p = new Put(row)
7869+
.addColumn(fam1, qual1, value1)
7870+
.addColumn(fam1, qual2, value2);
7871+
7872+
region.put(p);
7873+
7874+
Get get = new Get(row)
7875+
.addColumn(fam1, qual1)
7876+
.addColumn(fam1, qual2)
7877+
.setMaxResultSize(1); // 0 doesn't count as a limit, according to HBase
7878+
7879+
Result r = region.get(get);
7880+
7881+
assertTrue("Expected partial result, but result was not marked as partial", r.isPartial());
7882+
}
7883+
7884+
@Test
7885+
public void testGetsWithoutResultSizeLimitAreNotPartial() throws IOException {
7886+
HRegion region = initHRegion(tableName, name.getMethodName(), CONF, fam1);
7887+
7888+
Put p = new Put(row)
7889+
.addColumn(fam1, qual1, value1)
7890+
.addColumn(fam1, qual2, value2);
7891+
7892+
region.put(p);
7893+
7894+
Get get = new Get(row)
7895+
.addColumn(fam1, qual1)
7896+
.addColumn(fam1, qual2);
7897+
7898+
Result r = region.get(get);
7899+
7900+
assertFalse("Expected full result, but it was marked as partial", r.isPartial());
7901+
assertTrue(Bytes.equals(value1, r.getValue(fam1, qual1)));
7902+
assertTrue(Bytes.equals(value2, r.getValue(fam1, qual2)));
7903+
}
7904+
7905+
@Test
7906+
public void testGetsWithinResultSizeLimitAreNotPartial() throws IOException {
7907+
HRegion region = initHRegion(tableName, name.getMethodName(), CONF, fam1);
7908+
7909+
Put p = new Put(row)
7910+
.addColumn(fam1, qual1, value1)
7911+
.addColumn(fam1, qual2, value2);
7912+
7913+
region.put(p);
7914+
7915+
Get get = new Get(row)
7916+
.addColumn(fam1, qual1)
7917+
.addColumn(fam1, qual2)
7918+
.setMaxResultSize(Long.MAX_VALUE);
7919+
7920+
Result r = region.get(get);
7921+
7922+
assertFalse("Expected full result, but it was marked as partial", r.isPartial());
7923+
assertTrue(Bytes.equals(value1, r.getValue(fam1, qual1)));
7924+
assertTrue(Bytes.equals(value2, r.getValue(fam1, qual2)));
7925+
}
7926+
7927+
@Test
7928+
public void testGetsWithResultSizeLimitReturnPartialResults() throws IOException {
7929+
HRegion region = initHRegion(tableName, name.getMethodName(), CONF, fam1);
7930+
7931+
Put p = new Put(row)
7932+
.addColumn(fam1, qual1, value1)
7933+
.addColumn(fam1, qual2, value2);
7934+
7935+
region.put(p);
7936+
7937+
Get get = new Get(row)
7938+
.addColumn(fam1, qual1)
7939+
.addColumn(fam1, qual2)
7940+
.setMaxResultSize(10);
7941+
7942+
Result r = region.get(get);
7943+
7944+
assertTrue("Expected partial result, but it was marked as complete", r.mayHaveMoreCellsInRow());
7945+
assertTrue(Bytes.equals(value1, r.getValue(fam1, qual1)));
7946+
assertEquals("Got more results than expected", 1, r.size());
7947+
}
7948+
78647949
}

0 commit comments

Comments
 (0)