Skip to content

Commit 1d4cdc6

Browse files
committed
HBASE-8458 Support for batch version of checkAndMutate()
1 parent 4f9eecb commit 1d4cdc6

File tree

21 files changed

+2339
-276
lines changed

21 files changed

+2339
-276
lines changed

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBatchRpcRetryingCaller.java

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -256,22 +256,22 @@ private void failAll(Stream<Action> actions, int tries) {
256256
}
257257

258258
private ClientProtos.MultiRequest buildReq(Map<byte[], RegionRequest> actionsByRegion,
259-
List<CellScannable> cells, Map<Integer, Integer> rowMutationsIndexMap) throws IOException {
259+
List<CellScannable> cells, Map<Integer, Integer> indexMap) throws IOException {
260260
ClientProtos.MultiRequest.Builder multiRequestBuilder = ClientProtos.MultiRequest.newBuilder();
261261
ClientProtos.RegionAction.Builder regionActionBuilder = ClientProtos.RegionAction.newBuilder();
262262
ClientProtos.Action.Builder actionBuilder = ClientProtos.Action.newBuilder();
263263
ClientProtos.MutationProto.Builder mutationBuilder = ClientProtos.MutationProto.newBuilder();
264264
for (Map.Entry<byte[], RegionRequest> entry : actionsByRegion.entrySet()) {
265265
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
266266
// multiRequestBuilder will be populated with region actions.
267-
// rowMutationsIndexMap will be non-empty after the call if there is RowMutations in the
267+
// indexMap will be non-empty after the call if there is RowMutations/CheckAndMutate in the
268268
// action list.
269269
RequestConverter.buildNoDataRegionActions(entry.getKey(),
270270
entry.getValue().actions.stream()
271271
.sorted((a1, a2) -> Integer.compare(a1.getOriginalIndex(), a2.getOriginalIndex()))
272272
.collect(Collectors.toList()),
273-
cells, multiRequestBuilder, regionActionBuilder, actionBuilder, mutationBuilder, nonceGroup,
274-
rowMutationsIndexMap);
273+
cells, multiRequestBuilder, regionActionBuilder, actionBuilder, mutationBuilder,
274+
nonceGroup, indexMap);
275275
}
276276
return multiRequestBuilder.build();
277277
}
@@ -367,10 +367,10 @@ private void sendToServer(ServerName serverName, ServerRequest serverReq, int tr
367367
List<CellScannable> cells = new ArrayList<>();
368368
// Map from a created RegionAction to the original index for a RowMutations within
369369
// the original list of actions. This will be used to process the results when there
370-
// is RowMutations in the action list.
371-
Map<Integer, Integer> rowMutationsIndexMap = new HashMap<>();
370+
// is RowMutations/CheckAndMutate in the action list.
371+
Map<Integer, Integer> indexMap = new HashMap<>();
372372
try {
373-
req = buildReq(serverReq.actionsByRegion, cells, rowMutationsIndexMap);
373+
req = buildReq(serverReq.actionsByRegion, cells, indexMap);
374374
} catch (IOException e) {
375375
onError(serverReq.actionsByRegion, tries, e, serverName);
376376
return;
@@ -387,7 +387,7 @@ private void sendToServer(ServerName serverName, ServerRequest serverReq, int tr
387387
} else {
388388
try {
389389
onComplete(serverReq.actionsByRegion, tries, serverName, ResponseConverter.getResults(req,
390-
rowMutationsIndexMap, resp, controller.cellScanner()));
390+
indexMap, resp, controller.cellScanner()));
391391
} catch (Exception e) {
392392
onError(serverReq.actionsByRegion, tries, e, serverName);
393393
return;

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -231,12 +231,20 @@ default CompletableFuture<Long> incrementColumnValue(byte[] row, byte[] family,
231231
* });
232232
* </code>
233233
* </pre>
234+
*
235+
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
236+
* any more.
234237
*/
238+
@Deprecated
235239
CheckAndMutateBuilder checkAndMutate(byte[] row, byte[] family);
236240

237241
/**
238242
* A helper class for sending checkAndMutate request.
243+
*
244+
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
245+
* any more.
239246
*/
247+
@Deprecated
240248
interface CheckAndMutateBuilder {
241249

242250
/**
@@ -309,12 +317,20 @@ default CheckAndMutateBuilder ifEquals(byte[] value) {
309317
* });
310318
* </code>
311319
* </pre>
320+
*
321+
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
322+
* any more.
312323
*/
324+
@Deprecated
313325
CheckAndMutateWithFilterBuilder checkAndMutate(byte[] row, Filter filter);
314326

315327
/**
316328
* A helper class for sending checkAndMutate request with a filter.
329+
*
330+
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
331+
* any more.
317332
*/
333+
@Deprecated
318334
interface CheckAndMutateWithFilterBuilder {
319335

320336
/**
@@ -344,6 +360,35 @@ interface CheckAndMutateWithFilterBuilder {
344360
CompletableFuture<Boolean> thenMutate(RowMutations mutation);
345361
}
346362

363+
/**
364+
* checkAndMutate that atomically checks if a row matches the specified condition. If it does,
365+
* it performs the specified action.
366+
*
367+
* @param checkAndMutate The CheckAndMutate object.
368+
* @return A {@link CompletableFuture}s that represent the result for the CheckAndMutate.
369+
*/
370+
CompletableFuture<Boolean> checkAndMutate(CheckAndMutate checkAndMutate);
371+
372+
/**
373+
* Batch version of checkAndMutate.
374+
*
375+
* @param checkAndMutates The list of CheckAndMutate.
376+
* @return A list of {@link CompletableFuture}s that represent the result for each
377+
* CheckAndMutate.
378+
*/
379+
List<CompletableFuture<Boolean>> checkAndMutate(List<CheckAndMutate> checkAndMutates);
380+
381+
/**
382+
* A simple version of batch checkAndMutate. It will fail if there are any failures.
383+
*
384+
* @param checkAndMutates The list of rows to apply.
385+
* @return A {@link CompletableFuture} that wrapper the result boolean list.
386+
*/
387+
default CompletableFuture<List<Boolean>> checkAndMutateAll(
388+
List<CheckAndMutate> checkAndMutates) {
389+
return allOf(checkAndMutate(checkAndMutates));
390+
}
391+
347392
/**
348393
* Performs multiple mutations atomically on a single row. Currently {@link Put} and
349394
* {@link Delete} are supported.

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,17 @@ public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
205205
};
206206
}
207207

208+
@Override
209+
public CompletableFuture<Boolean> checkAndMutate(CheckAndMutate checkAndMutate) {
210+
return wrap(rawTable.checkAndMutate(checkAndMutate));
211+
}
212+
213+
@Override
214+
public List<CompletableFuture<Boolean>> checkAndMutate(List<CheckAndMutate> checkAndMutates) {
215+
return rawTable.checkAndMutate(checkAndMutates).stream()
216+
.map(this::wrap).collect(toList());
217+
}
218+
208219
@Override
209220
public CompletableFuture<Void> mutateRow(RowMutations mutation) {
210221
return wrap(rawTable.mutateRow(mutation));

0 commit comments

Comments
 (0)