You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by nd...@apache.org on 2024/09/17 13:18:49 UTC
(hbase) branch master updated: HBASE-28770: Support partial results in AggregateImplementation and AsyncAggregationClient (#6167)
This is an automated email from the ASF dual-hosted git repository.
ndimiduk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/master by this push:
new 5cf9170b8e1 HBASE-28770: Support partial results in AggregateImplementation and AsyncAggregationClient (#6167)
5cf9170b8e1 is described below
commit 5cf9170b8e1d5ac75d4018135b2afbe4803521b0
Author: Charles Connell <cc...@hubspot.com>
AuthorDate: Tue Sep 17 09:18:44 2024 -0400
HBASE-28770: Support partial results in AggregateImplementation and AsyncAggregationClient (#6167)
Signed-off-by: Duo Zhang <zh...@apache.org>
Signed-off-by: Nick Dimiduk <nd...@apache.org>
---
.../org/apache/hadoop/hbase/client/AsyncTable.java | 50 ++++
.../apache/hadoop/hbase/client/AsyncTableImpl.java | 95 ++++---
.../NoopPartialResultCoprocessorCallback.java | 62 +++++
.../hadoop/hbase/client/RawAsyncTableImpl.java | 86 ++++--
.../client/coprocessor/AggregationClient.java | 14 +-
.../client/coprocessor/AggregationHelper.java | 5 +-
.../client/coprocessor/AsyncAggregationClient.java | 290 +++++++++++++--------
.../hbase/coprocessor/AggregateImplementation.java | 173 ++++++++++--
.../MockPartialResultAggregateImplemention.java | 136 ++++++++++
.../TestAsyncAggregationClientPartialResult.java | 92 +++++++
.../protobuf/server/coprocessor/Aggregate.proto | 3 +
.../hadoop/hbase/client/DummyAsyncTable.java | 7 +
12 files changed, 818 insertions(+), 195 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java
index f14eac3cf79..200584581c4 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java
@@ -21,6 +21,7 @@ import static java.util.stream.Collectors.toList;
import static org.apache.hadoop.hbase.client.ConnectionUtils.toCheckExistenceOnly;
import static org.apache.hadoop.hbase.util.FutureUtils.allOf;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -676,6 +677,47 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
void onError(Throwable error);
}
+ /**
+ * Some coprocessors may support the idea of "partial results." If for some reason a coprocessor
+ * cannot return all results for a given region in a single response, the client side can be
+ * designed to recognize this and continuing requesting more results until they are completely
+ * accumulated in the client.
+ * <p>
+ * It is up to a particular coprocessor implementation and its corresponding clients to agree on
+ * what it means for results to be incomplete, how this state is communicated, and how multiple
+ * incomplete results are accumulated together.
+ * <p>
+ * Use this callback when you want to execute a coprocessor call on a range of regions, and that
+ * coprocessor may return incomplete results for a given region. See also the docs for
+ * {@link CoprocessorCallback}, which all apply here to its child interface too.
+ */
+ @InterfaceAudience.Public
+ interface PartialResultCoprocessorCallback<S, R> extends CoprocessorCallback<R> {
+ /**
+ * Subclasses should implement this to tell AsyncTable whether the given response is "final" or
+ * whether the AsyncTable should send another request to the coprocessor to fetch more results
+ * from the given region. This method of fetching more results can be used many times until
+ * there are no more results to fetch from the region.
+ * @param response The response received from the coprocessor
+ * @param region The region the response came from
+ * @return A ServiceCaller object if the response was not final and therefore another request is
+ * required to continuing fetching results. null if no more requests need to be sent to
+ * the region.
+ */
+ ServiceCaller<S, R> getNextCallable(R response, RegionInfo region);
+
+ /**
+ * Subclasses should implement this such that, when the above method returns non-null, this
+ * method returns the duration that AsyncTable should wait before sending the next request to
+ * the given region. You can use this to create a back-off behavior to reduce load on the
+ * RegionServer. If that's not desired, you can always return {@link Duration.ZERO}.
+ * @param response The response received from the coprocessor
+ * @param region The region the response came from
+ * @return The duration to wait.
+ */
+ Duration getWaitInterval(R response, RegionInfo region);
+ }
+
/**
* Helper class for sending coprocessorService request that executes a coprocessor call on regions
* which are covered by a range.
@@ -744,4 +786,12 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
*/
<S, R> CoprocessorServiceBuilder<S, R> coprocessorService(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback);
+
+ /**
+ * Similar to above. Use when your coprocessor client+endpoint supports partial results. If the
+ * server does not offer partial results, it is still safe to use this, assuming you implement
+ * your {@link PartialResultCoprocessorCallback#getNextCallable(Object, RegionInfo)} correctly.
+ */
+ <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(Function<RpcChannel, S> stubMaker,
+ ServiceCaller<S, R> callable, PartialResultCoprocessorCallback<S, R> callback);
}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java
index 3b411cea7fb..21765f77951 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java
@@ -23,6 +23,7 @@ import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import java.io.IOException;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -297,49 +298,69 @@ class AsyncTableImpl implements AsyncTable<ScanResultConsumer> {
public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
CoprocessorCallback<R> callback) {
+ return coprocessorService(stubMaker, callable,
+ new NoopPartialResultCoprocessorCallback<>(callback));
+ }
+
+ @Override
+ public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
+ Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
+ PartialResultCoprocessorCallback<S, R> callback) {
final Context context = Context.current();
- CoprocessorCallback<R> wrappedCallback = new CoprocessorCallback<R>() {
+ PartialResultCoprocessorCallback<S, R> wrappedCallback =
+ new PartialResultCoprocessorCallback<S, R>() {
+
+ private final Phaser regionCompletesInProgress = new Phaser(1);
+
+ @Override
+ public void onRegionComplete(RegionInfo region, R resp) {
+ regionCompletesInProgress.register();
+ pool.execute(context.wrap(() -> {
+ try {
+ callback.onRegionComplete(region, resp);
+ } finally {
+ regionCompletesInProgress.arriveAndDeregister();
+ }
+ }));
+ }
- private final Phaser regionCompletesInProgress = new Phaser(1);
+ @Override
+ public void onRegionError(RegionInfo region, Throwable error) {
+ regionCompletesInProgress.register();
+ pool.execute(context.wrap(() -> {
+ try {
+ callback.onRegionError(region, error);
+ } finally {
+ regionCompletesInProgress.arriveAndDeregister();
+ }
+ }));
+ }
- @Override
- public void onRegionComplete(RegionInfo region, R resp) {
- regionCompletesInProgress.register();
- pool.execute(context.wrap(() -> {
- try {
- callback.onRegionComplete(region, resp);
- } finally {
- regionCompletesInProgress.arriveAndDeregister();
- }
- }));
- }
+ @Override
+ public void onComplete() {
+ pool.execute(context.wrap(() -> {
+ // Guarantee that onComplete() is called after all onRegionComplete()'s are called
+ regionCompletesInProgress.arriveAndAwaitAdvance();
+ callback.onComplete();
+ }));
+ }
- @Override
- public void onRegionError(RegionInfo region, Throwable error) {
- regionCompletesInProgress.register();
- pool.execute(context.wrap(() -> {
- try {
- callback.onRegionError(region, error);
- } finally {
- regionCompletesInProgress.arriveAndDeregister();
- }
- }));
- }
+ @Override
+ public void onError(Throwable error) {
+ pool.execute(context.wrap(() -> callback.onError(error)));
+ }
- @Override
- public void onComplete() {
- pool.execute(context.wrap(() -> {
- // Guarantee that onComplete() is called after all onRegionComplete()'s are called
- regionCompletesInProgress.arriveAndAwaitAdvance();
- callback.onComplete();
- }));
- }
+ @Override
+ public ServiceCaller<S, R> getNextCallable(R response, RegionInfo region) {
+ return callback.getNextCallable(response, region);
+ }
- @Override
- public void onError(Throwable error) {
- pool.execute(context.wrap(() -> callback.onError(error)));
- }
- };
+ @Override
+ public Duration getWaitInterval(R response, RegionInfo region) {
+ return callback.getWaitInterval(response, region);
+ }
+
+ };
CoprocessorServiceBuilder<S, R> builder =
rawTable.coprocessorService(stubMaker, callable, wrappedCallback);
return new CoprocessorServiceBuilder<S, R>() {
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/NoopPartialResultCoprocessorCallback.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/NoopPartialResultCoprocessorCallback.java
new file mode 100644
index 00000000000..2218d361de2
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/NoopPartialResultCoprocessorCallback.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import java.time.Duration;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class NoopPartialResultCoprocessorCallback<S, R>
+ implements AsyncTable.PartialResultCoprocessorCallback<S, R> {
+
+ private final AsyncTable.CoprocessorCallback<R> delegate;
+
+ public NoopPartialResultCoprocessorCallback(AsyncTable.CoprocessorCallback<R> delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public void onRegionComplete(RegionInfo region, R resp) {
+ delegate.onRegionComplete(region, resp);
+ }
+
+ @Override
+ public void onRegionError(RegionInfo region, Throwable error) {
+ delegate.onRegionError(region, error);
+ }
+
+ @Override
+ public void onComplete() {
+ delegate.onComplete();
+ }
+
+ @Override
+ public void onError(Throwable error) {
+ delegate.onError(error);
+ }
+
+ @Override
+ public ServiceCaller<S, R> getNextCallable(R response, RegionInfo region) {
+ return null;
+ }
+
+ @Override
+ public Duration getWaitInterval(R response, RegionInfo region) {
+ return Duration.ZERO;
+ }
+}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java
index 257a5788b18..f2f00ed727d 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java
@@ -31,6 +31,7 @@ import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -791,10 +792,55 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
}
+ private <S, R> void coprocessorServiceUntilComplete(Function<RpcChannel, S> stubMaker,
+ ServiceCaller<S, R> callable, PartialResultCoprocessorCallback<S, R> callback,
+ AtomicBoolean locateFinished, AtomicInteger unfinishedRequest, RegionInfo region, Span span) {
+ addListener(coprocessorService(stubMaker, callable, region, region.getStartKey()), (r, e) -> {
+ try (Scope ignored = span.makeCurrent()) {
+ if (e != null) {
+ callback.onRegionError(region, e);
+ } else {
+ callback.onRegionComplete(region, r);
+ }
+
+ ServiceCaller<S, R> updatedCallable;
+ if (e == null && r != null) {
+ updatedCallable = callback.getNextCallable(r, region);
+ } else {
+ updatedCallable = null;
+ }
+
+ // If updatedCallable is non-null, we will be sending another request, so no need to
+ // decrement unfinishedRequest (recall that && short-circuits).
+ // If updatedCallable is null, and unfinishedRequest decrements to 0, we're done with the
+ // requests for this coprocessor call.
+ if (
+ updatedCallable == null && unfinishedRequest.decrementAndGet() == 0
+ && locateFinished.get()
+ ) {
+ callback.onComplete();
+ } else if (updatedCallable != null) {
+ Duration waitInterval = callback.getWaitInterval(r, region);
+ LOG.trace("Coprocessor returned incomplete result. "
+ + "Sleeping for {} before making follow-up request.", waitInterval);
+ if (waitInterval.isZero()) {
+ AsyncConnectionImpl.RETRY_TIMER.newTimeout(
+ (timeout) -> coprocessorServiceUntilComplete(stubMaker, updatedCallable, callback,
+ locateFinished, unfinishedRequest, region, span),
+ waitInterval.toMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ coprocessorServiceUntilComplete(stubMaker, updatedCallable, callback, locateFinished,
+ unfinishedRequest, region, span);
+ }
+ }
+ }
+ });
+ }
+
private <S, R> void onLocateComplete(Function<RpcChannel, S> stubMaker,
- ServiceCaller<S, R> callable, CoprocessorCallback<R> callback, List<HRegionLocation> locs,
- byte[] endKey, boolean endKeyInclusive, AtomicBoolean locateFinished,
- AtomicInteger unfinishedRequest, HRegionLocation loc, Throwable error) {
+ ServiceCaller<S, R> callable, PartialResultCoprocessorCallback<S, R> callback, byte[] endKey,
+ boolean endKeyInclusive, AtomicBoolean locateFinished, AtomicInteger unfinishedRequest,
+ HRegionLocation loc, Throwable error) {
final Span span = Span.current();
if (error != null) {
callback.onError(error);
@@ -810,23 +856,13 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
addListener(conn.getLocator().getRegionLocation(tableName, region.getEndKey(),
RegionLocateType.CURRENT, operationTimeoutNs), (l, e) -> {
try (Scope ignored = span.makeCurrent()) {
- onLocateComplete(stubMaker, callable, callback, locs, endKey, endKeyInclusive,
- locateFinished, unfinishedRequest, l, e);
+ onLocateComplete(stubMaker, callable, callback, endKey, endKeyInclusive, locateFinished,
+ unfinishedRequest, l, e);
}
});
}
- addListener(coprocessorService(stubMaker, callable, region, region.getStartKey()), (r, e) -> {
- try (Scope ignored = span.makeCurrent()) {
- if (e != null) {
- callback.onRegionError(region, e);
- } else {
- callback.onRegionComplete(region, r);
- }
- if (unfinishedRequest.decrementAndGet() == 0 && locateFinished.get()) {
- callback.onComplete();
- }
- }
- });
+ coprocessorServiceUntilComplete(stubMaker, callable, callback, locateFinished,
+ unfinishedRequest, region, span);
}
private final class CoprocessorServiceBuilderImpl<S, R>
@@ -836,7 +872,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
private final ServiceCaller<S, R> callable;
- private final CoprocessorCallback<R> callback;
+ private final PartialResultCoprocessorCallback<S, R> callback;
private byte[] startKey = HConstants.EMPTY_START_ROW;
@@ -847,7 +883,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
private boolean endKeyInclusive;
public CoprocessorServiceBuilderImpl(Function<RpcChannel, S> stubMaker,
- ServiceCaller<S, R> callable, CoprocessorCallback<R> callback) {
+ ServiceCaller<S, R> callable, PartialResultCoprocessorCallback<S, R> callback) {
this.stubMaker = Preconditions.checkNotNull(stubMaker, "stubMaker is null");
this.callable = Preconditions.checkNotNull(callable, "callable is null");
this.callback = Preconditions.checkNotNull(callback, "callback is null");
@@ -884,8 +920,8 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
.getRegionLocation(tableName, startKey, regionLocateType, operationTimeoutNs);
addListener(future, (loc, error) -> {
try (Scope ignored1 = span.makeCurrent()) {
- onLocateComplete(stubMaker, callable, callback, new ArrayList<>(), endKey,
- endKeyInclusive, new AtomicBoolean(false), new AtomicInteger(0), loc, error);
+ onLocateComplete(stubMaker, callable, callback, endKey, endKeyInclusive,
+ new AtomicBoolean(false), new AtomicInteger(0), loc, error);
}
});
}
@@ -896,6 +932,14 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
CoprocessorCallback<R> callback) {
+ return new CoprocessorServiceBuilderImpl<>(stubMaker, callable,
+ new NoopPartialResultCoprocessorCallback<>(callback));
+ }
+
+ @Override
+ public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
+ Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
+ PartialResultCoprocessorCallback<S, R> callback) {
return new CoprocessorServiceBuilderImpl<>(stubMaker, callable, callback);
}
}
diff --git a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java
index f47f5add0da..eb1ed795b55 100644
--- a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java
+++ b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java
@@ -227,7 +227,7 @@ public class AggregationClient implements Closeable {
*/
public <R, S, P extends Message, Q extends Message, T extends Message> R max(final Table table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
class MaxCallBack implements Batch.Callback<R> {
R max = null;
@@ -294,7 +294,7 @@ public class AggregationClient implements Closeable {
*/
public <R, S, P extends Message, Q extends Message, T extends Message> R min(final Table table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
class MinCallBack implements Batch.Callback<R> {
private R min = null;
@@ -370,7 +370,7 @@ public class AggregationClient implements Closeable {
public <R, S, P extends Message, Q extends Message, T extends Message> long
rowCount(final Table table, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, true);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, true, false);
class RowNumCallback implements Batch.Callback<Long> {
private final AtomicLong rowCountL = new AtomicLong(0);
@@ -436,7 +436,7 @@ public class AggregationClient implements Closeable {
*/
public <R, S, P extends Message, Q extends Message, T extends Message> S sum(final Table table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
class SumCallBack implements Batch.Callback<S> {
S sumVal = null;
@@ -505,7 +505,7 @@ public class AggregationClient implements Closeable {
private <R, S, P extends Message, Q extends Message, T extends Message> Pair<S, Long>
getAvgArgs(final Table table, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
class AvgCallBack implements Batch.Callback<Pair<S, Long>> {
S sum = null;
Long rowCount = 0L;
@@ -602,7 +602,7 @@ public class AggregationClient implements Closeable {
private <R, S, P extends Message, Q extends Message, T extends Message> Pair<List<S>, Long>
getStdArgs(final Table table, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
class StdCallback implements Batch.Callback<Pair<List<S>, Long>> {
long rowCountVal = 0L;
S sumVal = null, sumSqVal = null;
@@ -716,7 +716,7 @@ public class AggregationClient implements Closeable {
private <R, S, P extends Message, Q extends Message, T extends Message>
Pair<NavigableMap<byte[], List<S>>, List<S>> getMedianArgs(final Table table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
- final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
+ final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false, false);
final NavigableMap<byte[], List<S>> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
class StdCallback implements Batch.Callback<List<S>> {
S sumVal = null, sumWeights = null;
diff --git a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationHelper.java b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationHelper.java
index e7fa76ca274..8842e4d56a9 100644
--- a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationHelper.java
+++ b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationHelper.java
@@ -63,8 +63,8 @@ public final class AggregationHelper {
}
static <R, S, P extends Message, Q extends Message, T extends Message> AggregateRequest
- validateArgAndGetPB(Scan scan, ColumnInterpreter<R, S, P, Q, T> ci, boolean canFamilyBeAbsent)
- throws IOException {
+ validateArgAndGetPB(Scan scan, ColumnInterpreter<R, S, P, Q, T> ci, boolean canFamilyBeAbsent,
+ boolean supportPartialResults) throws IOException {
validateParameters(scan, canFamilyBeAbsent);
final AggregateRequest.Builder requestBuilder = AggregateRequest.newBuilder();
requestBuilder.setInterpreterClassName(ci.getClass().getCanonicalName());
@@ -73,6 +73,7 @@ public final class AggregationHelper {
requestBuilder.setInterpreterSpecificBytes(columnInterpreterSpecificData.toByteString());
}
requestBuilder.setScan(ProtobufUtil.toScan(scan));
+ requestBuilder.setClientSupportsPartialResult(supportPartialResults);
return requestBuilder.build();
}
diff --git a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AsyncAggregationClient.java b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AsyncAggregationClient.java
index 64bb187ad07..2faba432e87 100644
--- a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AsyncAggregationClient.java
+++ b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AsyncAggregationClient.java
@@ -22,26 +22,35 @@ import static org.apache.hadoop.hbase.client.coprocessor.AggregationHelper.valid
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
import java.util.Map;
import java.util.NavigableMap;
import java.util.NavigableSet;
import java.util.NoSuchElementException;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.AdvancedScanResultConsumer;
import org.apache.hadoop.hbase.client.AsyncTable;
-import org.apache.hadoop.hbase.client.AsyncTable.CoprocessorCallback;
+import org.apache.hadoop.hbase.client.AsyncTable.PartialResultCoprocessorCallback;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.ServiceCaller;
import org.apache.hadoop.hbase.coprocessor.ColumnInterpreter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.protobuf.Message;
+import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
+import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
+import org.apache.hbase.thirdparty.com.google.protobuf.TextFormat;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.AggregateRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.AggregateResponse;
@@ -54,12 +63,21 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.Aggrega
*/
@InterfaceAudience.Public
public final class AsyncAggregationClient {
+ private static final Logger log = LoggerFactory.getLogger(AsyncAggregationClient.class);
+
private AsyncAggregationClient() {
}
+ private interface StubCaller {
+ void call(AggregateService stub, RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> rpcCallback);
+ }
+
private static abstract class AbstractAggregationCallback<T>
- implements CoprocessorCallback<AggregateResponse> {
+ implements PartialResultCoprocessorCallback<AggregateService, AggregateResponse> {
private final CompletableFuture<T> future;
+ private final AggregateRequest originalRequest;
+ private final AsyncAggregationClient.StubCaller stubCaller;
protected boolean finished = false;
@@ -71,8 +89,11 @@ public final class AsyncAggregationClient {
future.completeExceptionally(error);
}
- protected AbstractAggregationCallback(CompletableFuture<T> future) {
+ protected AbstractAggregationCallback(CompletableFuture<T> future,
+ AggregateRequest originalRequest, AsyncAggregationClient.StubCaller stubCaller) {
this.future = future;
+ this.originalRequest = originalRequest;
+ this.stubCaller = stubCaller;
}
@Override
@@ -106,6 +127,31 @@ public final class AsyncAggregationClient {
finished = true;
future.complete(getFinalResult());
}
+
+ @Override
+ public ServiceCaller<AggregateService, AggregateResponse>
+ getNextCallable(AggregateResponse response, RegionInfo region) {
+ if (!response.hasNextChunkStartRow()) {
+ return null;
+ }
+ return (stub, controller, rpcCallback) -> {
+ AggregateRequest.Builder updatedRequest = AggregateRequest.newBuilder(originalRequest);
+ updatedRequest.setScan(originalRequest.getScan().toBuilder()
+ .setStartRow(response.getNextChunkStartRow()).build());
+ if (log.isTraceEnabled()) {
+ log.trace("Got incomplete result {} for original scan {}. Sending next request {}.",
+ TextFormat.shortDebugString(response),
+ TextFormat.shortDebugString(originalRequest.getScan()),
+ TextFormat.shortDebugString(updatedRequest));
+ }
+ stubCaller.call(stub, controller, updatedRequest.build(), rpcCallback);
+ };
+ }
+
+ @Override
+ public Duration getWaitInterval(AggregateResponse response, RegionInfo region) {
+ return Duration.ofMillis(response.getWaitIntervalMs());
+ }
}
private static <R, S, P extends Message, Q extends Message, T extends Message> R
@@ -131,32 +177,38 @@ public final class AsyncAggregationClient {
CompletableFuture<R> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<R> callback = new AbstractAggregationCallback<R>(future) {
+ AbstractAggregationCallback<R> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getMax) {
- private R max;
+ private R max;
+ private final Object lock = new Object();
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- if (resp.getFirstPartCount() > 0) {
- R result = getCellValueFromProto(ci, resp, 0);
- if (max == null || (result != null && ci.compare(max, result) < 0)) {
- max = result;
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ if (resp.getFirstPartCount() > 0) {
+ R result = getCellValueFromProto(ci, resp, 0);
+ synchronized (lock) {
+ if (max == null || (result != null && ci.compare(max, result) < 0)) {
+ max = result;
+ }
+ }
}
}
- }
- @Override
- protected R getFinalResult() {
- return max;
- }
- };
+ @Override
+ protected R getFinalResult() {
+ synchronized (lock) {
+ return max;
+ }
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getMax(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -168,33 +220,39 @@ public final class AsyncAggregationClient {
CompletableFuture<R> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<R> callback = new AbstractAggregationCallback<R>(future) {
+ AbstractAggregationCallback<R> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getMin) {
- private R min;
+ private R min;
+ private final Object lock = new Object();
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- if (resp.getFirstPartCount() > 0) {
- R result = getCellValueFromProto(ci, resp, 0);
- if (min == null || (result != null && ci.compare(min, result) > 0)) {
- min = result;
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ if (resp.getFirstPartCount() > 0) {
+ R result = getCellValueFromProto(ci, resp, 0);
+ synchronized (lock) {
+ if (min == null || (result != null && ci.compare(min, result) > 0)) {
+ min = result;
+ }
+ }
}
}
- }
- @Override
- protected R getFinalResult() {
- return min;
- }
- };
+ @Override
+ protected R getFinalResult() {
+ synchronized (lock) {
+ return min;
+ }
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getMin(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -207,27 +265,28 @@ public final class AsyncAggregationClient {
CompletableFuture<Long> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, true);
+ req = validateArgAndGetPB(scan, ci, true, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<Long> callback = new AbstractAggregationCallback<Long>(future) {
+ AbstractAggregationCallback<Long> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getRowNum) {
- private long count;
+ private final AtomicLong count = new AtomicLong(0L);
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- count += resp.getFirstPart(0).asReadOnlyByteBuffer().getLong();
- }
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ count.addAndGet(resp.getFirstPart(0).asReadOnlyByteBuffer().getLong());
+ }
- @Override
- protected Long getFinalResult() {
- return count;
- }
- };
+ @Override
+ protected Long getFinalResult() {
+ return count.get();
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getRowNum(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -239,29 +298,35 @@ public final class AsyncAggregationClient {
CompletableFuture<S> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<S> callback = new AbstractAggregationCallback<S>(future) {
- private S sum;
+ AbstractAggregationCallback<S> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getSum) {
+ private S sum;
+ private final Object lock = new Object();
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- if (resp.getFirstPartCount() > 0) {
- S s = getPromotedValueFromProto(ci, resp, 0);
- sum = ci.add(sum, s);
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ if (resp.getFirstPartCount() > 0) {
+ S s = getPromotedValueFromProto(ci, resp, 0);
+ synchronized (lock) {
+ sum = ci.add(sum, s);
+ }
+ }
}
- }
- @Override
- protected S getFinalResult() {
- return sum;
- }
- };
+ @Override
+ protected S getFinalResult() {
+ synchronized (lock) {
+ return sum;
+ }
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getSum(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -274,31 +339,36 @@ public final class AsyncAggregationClient {
CompletableFuture<Double> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<Double> callback = new AbstractAggregationCallback<Double>(future) {
- private S sum;
-
- long count = 0L;
+ AbstractAggregationCallback<Double> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getAvg) {
+ private S sum;
+ long count = 0L;
+ private final Object lock = new Object();
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- if (resp.getFirstPartCount() > 0) {
- sum = ci.add(sum, getPromotedValueFromProto(ci, resp, 0));
- count += resp.getSecondPart().asReadOnlyByteBuffer().getLong();
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ if (resp.getFirstPartCount() > 0) {
+ synchronized (lock) {
+ sum = ci.add(sum, getPromotedValueFromProto(ci, resp, 0));
+ count += resp.getSecondPart().asReadOnlyByteBuffer().getLong();
+ }
+ }
}
- }
- @Override
- protected Double getFinalResult() {
- return ci.divideForAvg(sum, count);
- }
- };
+ @Override
+ protected Double getFinalResult() {
+ synchronized (lock) {
+ return ci.divideForAvg(sum, count);
+ }
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getAvg(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -311,37 +381,41 @@ public final class AsyncAggregationClient {
CompletableFuture<Double> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
- AbstractAggregationCallback<Double> callback = new AbstractAggregationCallback<Double>(future) {
-
- private S sum;
-
- private S sumSq;
+ AbstractAggregationCallback<Double> callback =
+ new AbstractAggregationCallback<>(future, req, AggregateService::getStd) {
- private long count;
+ private S sum;
+ private S sumSq;
+ private long count;
+ private final Object lock = new Object();
- @Override
- protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
- if (resp.getFirstPartCount() > 0) {
- sum = ci.add(sum, getPromotedValueFromProto(ci, resp, 0));
- sumSq = ci.add(sumSq, getPromotedValueFromProto(ci, resp, 1));
- count += resp.getSecondPart().asReadOnlyByteBuffer().getLong();
+ @Override
+ protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
+ if (resp.getFirstPartCount() > 0) {
+ synchronized (lock) {
+ sum = ci.add(sum, getPromotedValueFromProto(ci, resp, 0));
+ sumSq = ci.add(sumSq, getPromotedValueFromProto(ci, resp, 1));
+ count += resp.getSecondPart().asReadOnlyByteBuffer().getLong();
+ }
+ }
}
- }
- @Override
- protected Double getFinalResult() {
- double avg = ci.divideForAvg(sum, count);
- double avgSq = ci.divideForAvg(sumSq, count);
- return Math.sqrt(avgSq - avg * avg);
- }
- };
+ @Override
+ protected Double getFinalResult() {
+ synchronized (lock) {
+ double avg = ci.divideForAvg(sum, count);
+ double avgSq = ci.divideForAvg(sumSq, count);
+ return Math.sqrt(avgSq - avg * avg);
+ }
+ }
+ };
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getStd(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
@@ -352,20 +426,20 @@ public final class AsyncAggregationClient {
private static <R, S, P extends Message, Q extends Message, T extends Message>
CompletableFuture<NavigableMap<byte[], S>>
sumByRegion(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
- CompletableFuture<NavigableMap<byte[], S>> future =
- new CompletableFuture<NavigableMap<byte[], S>>();
+ CompletableFuture<NavigableMap<byte[], S>> future = new CompletableFuture<>();
AggregateRequest req;
try {
- req = validateArgAndGetPB(scan, ci, false);
+ req = validateArgAndGetPB(scan, ci, false, true);
} catch (IOException e) {
future.completeExceptionally(e);
return future;
}
int firstPartIndex = scan.getFamilyMap().get(scan.getFamilies()[0]).size() - 1;
AbstractAggregationCallback<NavigableMap<byte[], S>> callback =
- new AbstractAggregationCallback<NavigableMap<byte[], S>>(future) {
+ new AbstractAggregationCallback<>(future, req, AggregateService::getMedian) {
- private final NavigableMap<byte[], S> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ private final NavigableMap<byte[], S> map =
+ Collections.synchronizedNavigableMap(new TreeMap<>(Bytes.BYTES_COMPARATOR));
@Override
protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
@@ -380,7 +454,7 @@ public final class AsyncAggregationClient {
}
};
table
- .<AggregateService, AggregateResponse> coprocessorService(AggregateService::newStub,
+ .coprocessorService(AggregateService::newStub,
(stub, controller, rpcCallback) -> stub.getMedian(controller, req, rpcCallback), callback)
.fromRow(nullToEmpty(scan.getStartRow()), scan.includeStartRow())
.toRow(nullToEmpty(scan.getStopRow()), scan.includeStopRow()).execute();
diff --git a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java
index 6adb9fc22eb..39bfaae5a65 100644
--- a/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java
+++ b/hbase-endpoint/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java
@@ -23,15 +23,20 @@ import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.NavigableSet;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.client.ClientUtil;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
+import org.apache.hadoop.hbase.ipc.RpcServer;
+import org.apache.hadoop.hbase.quotas.OperationQuota;
+import org.apache.hadoop.hbase.quotas.RpcThrottlingException;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
@@ -77,7 +82,9 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
InternalScanner scanner = null;
AggregateResponse response = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
T max = null;
+ boolean hasMoreRows = false;
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
T temp;
@@ -86,24 +93,28 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
List<Cell> results = new ArrayList<>();
byte[] colFamily = scan.getFamilies()[0];
NavigableSet<byte[]> qualifiers = scan.getFamilyMap().get(colFamily);
+ // qualifier can be null.
byte[] qualifier = null;
if (qualifiers != null && !qualifiers.isEmpty()) {
qualifier = qualifiers.pollFirst();
}
- // qualifier can be null.
- boolean hasMoreRows = false;
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
hasMoreRows = scanner.next(results);
int listSize = results.size();
for (int i = 0; i < listSize; i++) {
temp = ci.getValue(colFamily, qualifier, results.get(i));
max = (max == null || (temp != null && ci.compare(temp, max) > 0)) ? temp : max;
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
} while (hasMoreRows);
if (max != null) {
AggregateResponse.Builder builder = AggregateResponse.newBuilder();
builder.addFirstPart(ci.getProtoForCellType(max).toByteString());
+ setPartialResultResponse(builder, request, hasMoreRows, partialResultContext);
response = builder.build();
}
} catch (IOException e) {
@@ -112,9 +123,11 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
- log.info("Maximum from this region is "
- + env.getRegion().getRegionInfo().getRegionNameAsString() + ": " + max);
+ log.debug("Maximum from this region is {}: {} (partial result: {}) (client {})",
+ env.getRegion().getRegionInfo().getRegionNameAsString(), max, hasMoreRows,
+ RpcServer.getRequestUser());
done.run(response);
}
@@ -129,7 +142,9 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
AggregateResponse response = null;
InternalScanner scanner = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
T min = null;
+ boolean hasMoreRows = false;
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
T temp;
@@ -142,19 +157,24 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (qualifiers != null && !qualifiers.isEmpty()) {
qualifier = qualifiers.pollFirst();
}
- boolean hasMoreRows = false;
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
hasMoreRows = scanner.next(results);
int listSize = results.size();
for (int i = 0; i < listSize; i++) {
temp = ci.getValue(colFamily, qualifier, results.get(i));
min = (min == null || (temp != null && ci.compare(temp, min) < 0)) ? temp : min;
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
} while (hasMoreRows);
if (min != null) {
- response = AggregateResponse.newBuilder()
- .addFirstPart(ci.getProtoForCellType(min).toByteString()).build();
+ AggregateResponse.Builder responseBuilder =
+ AggregateResponse.newBuilder().addFirstPart(ci.getProtoForCellType(min).toByteString());
+ setPartialResultResponse(responseBuilder, request, hasMoreRows, partialResultContext);
+ response = responseBuilder.build();
}
} catch (IOException e) {
CoprocessorRpcUtils.setControllerException(controller, e);
@@ -162,9 +182,11 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
- log.info("Minimum from this region is "
- + env.getRegion().getRegionInfo().getRegionNameAsString() + ": " + min);
+ log.debug("Minimum from this region is {}: {} (partial result: {}) (client {})",
+ env.getRegion().getRegionInfo().getRegionNameAsString(), min, hasMoreRows,
+ RpcServer.getRequestUser());
done.run(response);
}
@@ -179,7 +201,9 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
AggregateResponse response = null;
InternalScanner scanner = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
long sum = 0L;
+ boolean hasMoreRows = false;
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null;
@@ -193,8 +217,10 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
qualifier = qualifiers.pollFirst();
}
List<Cell> results = new ArrayList<>();
- boolean hasMoreRows = false;
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
hasMoreRows = scanner.next(results);
int listSize = results.size();
for (int i = 0; i < listSize; i++) {
@@ -203,11 +229,14 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
sumVal = ci.add(sumVal, ci.castToReturnType(temp));
}
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
} while (hasMoreRows);
if (sumVal != null) {
- response = AggregateResponse.newBuilder()
- .addFirstPart(ci.getProtoForPromotedType(sumVal).toByteString()).build();
+ AggregateResponse.Builder responseBuilder = AggregateResponse.newBuilder()
+ .addFirstPart(ci.getProtoForPromotedType(sumVal).toByteString());
+ setPartialResultResponse(responseBuilder, request, hasMoreRows, partialResultContext);
+ response = responseBuilder.build();
}
} catch (IOException e) {
CoprocessorRpcUtils.setControllerException(controller, e);
@@ -215,9 +244,11 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
- log.debug("Sum from this region is " + env.getRegion().getRegionInfo().getRegionNameAsString()
- + ": " + sum);
+ log.debug("Sum from this region is {}: {} (partial result: {}) (client {})",
+ env.getRegion().getRegionInfo().getRegionNameAsString(), sum, hasMoreRows,
+ RpcServer.getRequestUser());
done.run(response);
}
@@ -232,6 +263,8 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
long counter = 0L;
List<Cell> results = new ArrayList<>();
InternalScanner scanner = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
+ boolean hasMoreRows = false;
try {
Scan scan = ProtobufUtil.toScan(request.getScan());
byte[][] colFamilies = scan.getFamilies();
@@ -246,26 +279,34 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
scan.setFilter(new FirstKeyOnlyFilter());
}
scanner = env.getRegion().getScanner(scan);
- boolean hasMoreRows = false;
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
hasMoreRows = scanner.next(results);
- if (results.size() > 0) {
+ if (!results.isEmpty()) {
counter++;
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
} while (hasMoreRows);
ByteBuffer bb = ByteBuffer.allocate(8).putLong(counter);
bb.rewind();
- response = AggregateResponse.newBuilder().addFirstPart(ByteString.copyFrom(bb)).build();
+ AggregateResponse.Builder responseBuilder =
+ AggregateResponse.newBuilder().addFirstPart(ByteString.copyFrom(bb));
+ setPartialResultResponse(responseBuilder, request, hasMoreRows, partialResultContext);
+ response = responseBuilder.build();
} catch (IOException e) {
CoprocessorRpcUtils.setControllerException(controller, e);
} finally {
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
- log.info("Row counter from this region is "
- + env.getRegion().getRegionInfo().getRegionNameAsString() + ": " + counter);
+ log.debug("Row counter from this region is {}: {} (partial result: {}) (client {})",
+ env.getRegion().getRegionInfo().getRegionNameAsString(), counter, hasMoreRows,
+ RpcServer.getRequestUser());
done.run(response);
}
@@ -284,6 +325,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
AggregateResponse response = null;
InternalScanner scanner = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null;
@@ -301,6 +343,9 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
do {
results.clear();
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
hasMoreRows = scanner.next(results);
int listSize = results.size();
for (int i = 0; i < listSize; i++) {
@@ -308,6 +353,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
ci.add(sumVal, ci.castToReturnType(ci.getValue(colFamily, qualifier, results.get(i))));
}
rowCountVal++;
+ postScanPartialResultUpdate(results, partialResultContext);
} while (hasMoreRows);
if (sumVal != null) {
ByteString first = ci.getProtoForPromotedType(sumVal).toByteString();
@@ -316,6 +362,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
ByteBuffer bb = ByteBuffer.allocate(8).putLong(rowCountVal);
bb.rewind();
pair.setSecondPart(ByteString.copyFrom(bb));
+ setPartialResultResponse(pair, request, hasMoreRows, partialResultContext);
response = pair.build();
}
} catch (IOException e) {
@@ -324,6 +371,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
done.run(response);
}
@@ -341,6 +389,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
InternalScanner scanner = null;
AggregateResponse response = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null, sumSqVal = null, tempVal = null;
@@ -358,6 +407,9 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
boolean hasMoreRows = false;
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
tempVal = null;
hasMoreRows = scanner.next(results);
int listSize = results.size();
@@ -365,6 +417,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
tempVal =
ci.add(tempVal, ci.castToReturnType(ci.getValue(colFamily, qualifier, results.get(i))));
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
sumVal = ci.add(sumVal, tempVal);
sumSqVal = ci.add(sumSqVal, ci.multiply(tempVal, tempVal));
@@ -379,6 +432,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
ByteBuffer bb = ByteBuffer.allocate(8).putLong(rowCountVal);
bb.rewind();
pair.setSecondPart(ByteString.copyFrom(bb));
+ setPartialResultResponse(pair, request, hasMoreRows, partialResultContext);
response = pair.build();
}
} catch (IOException e) {
@@ -387,6 +441,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
done.run(response);
}
@@ -402,6 +457,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
RpcCallback<AggregateResponse> done) {
AggregateResponse response = null;
InternalScanner scanner = null;
+ PartialResultContext partialResultContext = new PartialResultContext();
try {
ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null, sumWeights = null, tempVal = null, tempWeight = null;
@@ -418,8 +474,10 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
List<Cell> results = new ArrayList<>();
boolean hasMoreRows = false;
-
do {
+ if (shouldBreakForThrottling(request, scan, partialResultContext)) {
+ break;
+ }
tempVal = null;
tempWeight = null;
hasMoreRows = scanner.next(results);
@@ -432,6 +490,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
ci.add(tempWeight, ci.castToReturnType(ci.getValue(colFamily, weightQualifier, kv)));
}
}
+ postScanPartialResultUpdate(results, partialResultContext);
results.clear();
sumVal = ci.add(sumVal, tempVal);
sumWeights = ci.add(sumWeights, tempWeight);
@@ -442,6 +501,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
AggregateResponse.Builder pair = AggregateResponse.newBuilder();
pair.addFirstPart(first_sumVal);
pair.addFirstPart(first_sumWeights);
+ setPartialResultResponse(pair, request, hasMoreRows, partialResultContext);
response = pair.build();
} catch (IOException e) {
CoprocessorRpcUtils.setControllerException(controller, e);
@@ -449,10 +509,73 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
done.run(response);
}
+ private final static class PartialResultContext {
+ private OperationQuota quota = null;
+ private long waitIntervalMs = 0;
+ private byte[] lastRowSuccessfullyProcessedArray = null;
+ private int lastRowSuccessfullyProcessedOffset = 0;
+ private int lastRowSuccessfullyProcessedLength = 0;
+ private long previousReadConsumed = 0;
+ private long previousReadConsumedDifference = 0;
+ }
+
+ private boolean shouldBreakForThrottling(AggregateRequest request, Scan scan,
+ PartialResultContext context) throws IOException {
+ if (request.getClientSupportsPartialResult()) {
+ long maxBlockBytesScanned;
+ if (context.quota == null) {
+ maxBlockBytesScanned = Long.MAX_VALUE;
+ } else {
+ maxBlockBytesScanned = context.quota.getMaxResultSize();
+ }
+ try {
+ context.quota =
+ env.checkScanQuota(scan, maxBlockBytesScanned, context.previousReadConsumedDifference);
+ } catch (RpcThrottlingException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("Ending early for throttling for region {}",
+ env.getRegion().getRegionInfo().getRegionNameAsString());
+ }
+ context.waitIntervalMs = e.getWaitInterval();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void postScanPartialResultUpdate(List<Cell> results, PartialResultContext context) {
+ if (context.quota != null) {
+ context.quota.addScanResultCells(results);
+ }
+ if (!results.isEmpty()) {
+ Cell result = results.get(results.size() - 1);
+ context.lastRowSuccessfullyProcessedArray = result.getRowArray();
+ context.lastRowSuccessfullyProcessedOffset = result.getRowOffset();
+ context.lastRowSuccessfullyProcessedLength = result.getRowLength();
+ }
+ }
+
+ private void setPartialResultResponse(AggregateResponse.Builder builder, AggregateRequest request,
+ boolean hasMoreRows, PartialResultContext context) throws IOException {
+ // If we encountered an RpcThrottlingException, tell the client the partial result we've
+ // accumulated so far, and what row to start scanning at in order to finish the scan.
+ if (request.getClientSupportsPartialResult() && hasMoreRows) {
+ if (context.lastRowSuccessfullyProcessedArray != null) {
+ byte[] lastRowSuccessfullyProcessed = Arrays.copyOfRange(
+ context.lastRowSuccessfullyProcessedArray, context.lastRowSuccessfullyProcessedOffset,
+ context.lastRowSuccessfullyProcessedOffset + context.lastRowSuccessfullyProcessedLength);
+ builder.setNextChunkStartRow(ByteString.copyFrom(
+ ClientUtil.calculateTheClosestNextRowKeyForPrefix(lastRowSuccessfullyProcessed)));
+ }
+ builder.setWaitIntervalMs(context.waitIntervalMs);
+ }
+ }
+
@SuppressWarnings("unchecked")
// Used server-side too by Aggregation Coprocesor Endpoint. Undo this interdependence. TODO.
ColumnInterpreter<T, S, P, Q, R> constructColumnInterpreterFromRequest(AggregateRequest request)
@@ -503,4 +626,14 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
public void stop(CoprocessorEnvironment env) throws IOException {
// nothing to do
}
+
+ private void closeQuota(PartialResultContext context) {
+ if (context != null && context.quota != null) {
+ context.quota.close();
+ long readConsumed = context.quota.getReadConsumed();
+ context.previousReadConsumedDifference = readConsumed - context.previousReadConsumed;
+ context.previousReadConsumed = readConsumed;
+ }
+ }
+
}
diff --git a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/MockPartialResultAggregateImplemention.java b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/MockPartialResultAggregateImplemention.java
new file mode 100644
index 00000000000..a36f5889c9c
--- /dev/null
+++ b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/MockPartialResultAggregateImplemention.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
+import org.apache.hbase.thirdparty.com.google.protobuf.Message;
+import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
+import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
+import org.apache.hbase.thirdparty.com.google.protobuf.Service;
+
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.AggregateRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.AggregateResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AggregateProtos.AggregateService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+
+@InterfaceAudience.Private
+public class MockPartialResultAggregateImplemention<T, S, P extends Message, Q extends Message,
+ R extends Message> extends AggregateService implements RegionCoprocessor {
+ private RegionCoprocessorEnvironment env;
+ private ConcurrentMap<String, Boolean> seenRegion;
+
+ /**
+ * Mocks the behavior of getMax() in the real AggregateImplementation. Makes some fake data to
+ * send back some partial results, and then finally a complete result.
+ */
+ @Override
+ public void getMax(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+ AggregateResponse.Builder result = AggregateResponse.newBuilder();
+
+ int max;
+ boolean seenRegionBefore = seenRegion.put(env.getRegionInfo().getEncodedName(), true) != null;
+ if (seenRegionBefore) {
+ max = 10;
+ } else {
+ max = 2;
+ seenRegion.put(env.getRegionInfo().getEncodedName(), true);
+ }
+ HBaseProtos.LongMsg longMsg = HBaseProtos.LongMsg.newBuilder().setLongMsg(max).build();
+ result.addFirstPart(longMsg.toByteString());
+ if (!seenRegionBefore) {
+ result.setNextChunkStartRow(ByteString.copyFrom(env.getRegionInfo().getStartKey()));
+ }
+ done.run(result.build());
+ }
+
+ /**
+ * Mocks the behavior of getMin() in the real AggregateImplementation. Makes some fake data to
+ * send back some partial results, and then finally a complete result.
+ */
+ @Override
+ public void getMin(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+ AggregateResponse.Builder result = AggregateResponse.newBuilder();
+ int min;
+ boolean seenRegionBefore = seenRegion.put(env.getRegionInfo().getEncodedName(), true) != null;
+ if (seenRegionBefore) {
+ min = 10;
+ } else {
+ min = 2;
+ seenRegion.put(env.getRegionInfo().getEncodedName(), true);
+ }
+ HBaseProtos.LongMsg longMsg = HBaseProtos.LongMsg.newBuilder().setLongMsg(min).build();
+ result.addFirstPart(longMsg.toByteString());
+ if (!seenRegionBefore) {
+ result.setNextChunkStartRow(ByteString.copyFrom(env.getRegionInfo().getStartKey()));
+ }
+ done.run(result.build());
+ }
+
+ @Override
+ public void getSum(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+
+ }
+
+ @Override
+ public void getRowNum(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+
+ }
+
+ @Override
+ public void getAvg(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+
+ }
+
+ @Override
+ public void getStd(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+
+ }
+
+ @Override
+ public void getMedian(RpcController controller, AggregateRequest request,
+ RpcCallback<AggregateResponse> done) {
+
+ }
+
+ @Override
+ public Iterable<Service> getServices() {
+ return Collections.singleton(this);
+ }
+
+ @Override
+ public void start(CoprocessorEnvironment env) throws IOException {
+ this.env = (RegionCoprocessorEnvironment) env;
+ this.seenRegion = new ConcurrentHashMap<>();
+ }
+
+}
diff --git a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAggregationClientPartialResult.java b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAggregationClientPartialResult.java
new file mode 100644
index 00000000000..6b6909d49df
--- /dev/null
+++ b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAggregationClientPartialResult.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.ExecutionException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.coprocessor.AsyncAggregationClient;
+import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ MediumTests.class, CoprocessorTests.class })
+public class TestAsyncAggregationClientPartialResult {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestAsyncAggregationClientPartialResult.class);
+
+ private static HBaseTestingUtil UTIL = new HBaseTestingUtil();
+
+ private static TableName TABLE_NAME =
+ TableName.valueOf("TestAsyncAggregationClientPartialResult");
+
+ private static byte[] CF = Bytes.toBytes("CF");
+
+ private static byte[] CQ = Bytes.toBytes("CQ");
+
+ private static AsyncConnection CONN;
+
+ private static AsyncTable<AdvancedScanResultConsumer> TABLE;
+
+ @Before
+ public void setUp() throws Exception {
+ Configuration conf = UTIL.getConfiguration();
+ conf.setStrings(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
+ MockPartialResultAggregateImplemention.class.getName());
+ UTIL.startMiniCluster(3);
+ byte[][] splitKeys = new byte[8][];
+ for (int i = 111; i < 999; i += 111) {
+ splitKeys[i / 111 - 1] = Bytes.toBytes(String.format("%03d", i));
+ }
+ UTIL.createTable(TABLE_NAME, CF, splitKeys);
+ CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
+ TABLE = CONN.getTable(TABLE_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CONN.close();
+ UTIL.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testMax() throws InterruptedException, ExecutionException {
+ assertEquals(10, AsyncAggregationClient
+ .max(TABLE, new LongColumnInterpreter(), new Scan().addColumn(CF, CQ)).get().longValue());
+ }
+
+ @Test
+ public void testMin() throws InterruptedException, ExecutionException {
+ assertEquals(2, AsyncAggregationClient
+ .min(TABLE, new LongColumnInterpreter(), new Scan().addColumn(CF, CQ)).get().longValue());
+ }
+
+}
diff --git a/hbase-protocol-shaded/src/main/protobuf/server/coprocessor/Aggregate.proto b/hbase-protocol-shaded/src/main/protobuf/server/coprocessor/Aggregate.proto
index f76eff5fcfa..5b411d85163 100644
--- a/hbase-protocol-shaded/src/main/protobuf/server/coprocessor/Aggregate.proto
+++ b/hbase-protocol-shaded/src/main/protobuf/server/coprocessor/Aggregate.proto
@@ -36,6 +36,7 @@ message AggregateRequest {
required string interpreter_class_name = 1;
required Scan scan = 2;
optional bytes interpreter_specific_bytes = 3;
+ optional bool client_supports_partial_result = 4 [default = false];
}
message AggregateResponse {
@@ -48,6 +49,8 @@ message AggregateResponse {
*/
repeated bytes first_part = 1;
optional bytes second_part = 2;
+ optional bytes next_chunk_start_row = 3;
+ optional uint64 wait_interval_ms = 4 [default = 0];
}
/** Refer to the AggregateImplementation class for an overview of the
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/DummyAsyncTable.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/DummyAsyncTable.java
index 45e59def721..27787bd62db 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/DummyAsyncTable.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/DummyAsyncTable.java
@@ -180,4 +180,11 @@ public class DummyAsyncTable<C extends ScanResultConsumerBase> implements AsyncT
CoprocessorCallback<R> callback) {
return null;
}
+
+ @Override
+ public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
+ Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
+ PartialResultCoprocessorCallback<S, R> callback) {
+ return null;
+ }
}