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/23 09:18:43 UTC
(hbase) branch branch-2 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 branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2 by this push:
new 5c0c27d961d HBASE-28770: Support partial results in AggregateImplementation and AsyncAggregationClient (#6167)
5c0c27d961d is described below
commit 5c0c27d961db1482ed6bfa8f48a7b709a087878d
Author: Charles Connell <cc...@hubspot.com>
AuthorDate: Mon Sep 23 05:18:37 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 | 291 +++++++++++++--------
.../hbase/coprocessor/AggregateImplementation.java | 173 ++++++++++--
hbase-endpoint/src/main/protobuf/Aggregate.proto | 3 +
.../MockPartialResultAggregateImplemention.java | 134 ++++++++++
.../TestAsyncAggregationClientPartialResult.java | 92 +++++++
.../hadoop/hbase/client/DummyAsyncTable.java | 6 +
12 files changed, 816 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 d6117dbd3b0..9cc9b56829b 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
@@ -22,6 +22,7 @@ import static org.apache.hadoop.hbase.client.ConnectionUtils.toCheckExistenceOnl
import static org.apache.hadoop.hbase.util.FutureUtils.allOf;
import com.google.protobuf.RpcChannel;
+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 e19ed207c24..53764f7790d 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
@@ -24,6 +24,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;
@@ -296,49 +297,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 a5dc77d6e8e..6f22b35664b 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
@@ -32,6 +32,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;
@@ -829,10 +830,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);
@@ -848,23 +894,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>
@@ -874,7 +910,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;
@@ -885,7 +921,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");
@@ -922,8 +958,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);
}
});
}
@@ -934,6 +970,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 59d89a4f63a..8cdbb12cc1f 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
@@ -225,7 +225,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;
@@ -292,7 +292,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;
@@ -368,7 +368,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);
@@ -434,7 +434,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;
@@ -503,7 +503,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;
@@ -600,7 +600,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;
@@ -714,7 +714,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 d1d11fec5c3..5519a7e75c2 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
@@ -61,8 +61,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());
@@ -71,6 +71,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 b477fbef988..8cb836afe3f 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,21 +22,28 @@ import static org.apache.hadoop.hbase.client.coprocessor.AggregationHelper.valid
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import com.google.protobuf.Message;
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.TextFormat;
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.protobuf.generated.AggregateProtos.AggregateRequest;
import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateResponse;
@@ -44,6 +51,8 @@ import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateServi
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;
/**
* This client class is for invoking the aggregate functions deployed on the Region Server side via
@@ -52,12 +61,21 @@ import org.apache.yetus.audience.InterfaceAudience;
*/
@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;
@@ -69,8 +87,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
@@ -104,6 +125,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
@@ -129,32 +175,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<R>(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();
@@ -166,33 +218,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<R>(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();
@@ -205,27 +263,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<Long>(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();
@@ -237,29 +296,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<S>(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();
@@ -272,31 +337,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<Double>(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();
@@ -309,37 +379,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<Double>(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();
@@ -350,20 +424,21 @@ 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<NavigableMap<byte[], S>>(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 {
@@ -378,7 +453,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 bea7291b576..770a338481f 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
@@ -28,19 +28,24 @@ 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.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateRequest;
import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateResponse;
import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService;
+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;
@@ -75,7 +80,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;
@@ -84,24 +91,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) {
@@ -110,9 +121,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);
}
@@ -127,7 +140,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;
@@ -140,19 +155,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);
@@ -160,9 +180,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);
}
@@ -177,7 +199,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;
@@ -191,8 +215,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++) {
@@ -201,11 +227,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);
@@ -213,9 +242,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);
}
@@ -230,6 +261,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();
@@ -244,26 +277,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);
}
@@ -282,6 +323,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;
@@ -299,6 +341,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++) {
@@ -306,6 +351,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();
@@ -314,6 +360,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) {
@@ -322,6 +369,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
done.run(response);
}
@@ -339,6 +387,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;
@@ -356,6 +405,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();
@@ -363,6 +415,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));
@@ -377,6 +430,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) {
@@ -385,6 +439,7 @@ public class AggregateImplementation<T, S, P extends Message, Q extends Message,
if (scanner != null) {
IOUtils.closeQuietly(scanner);
}
+ closeQuota(partialResultContext);
}
done.run(response);
}
@@ -400,6 +455,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;
@@ -416,8 +472,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);
@@ -430,6 +488,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);
@@ -440,6 +499,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);
@@ -447,10 +507,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)
@@ -501,4 +624,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/main/protobuf/Aggregate.proto b/hbase-endpoint/src/main/protobuf/Aggregate.proto
index bf1adcb45cd..469bab011c9 100644
--- a/hbase-endpoint/src/main/protobuf/Aggregate.proto
+++ b/hbase-endpoint/src/main/protobuf/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-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..9006c2999c0
--- /dev/null
+++ b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/MockPartialResultAggregateImplemention.java
@@ -0,0 +1,134 @@
+/*
+ * 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 com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+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.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateRequest;
+import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateResponse;
+import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@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..f199e216323
--- /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.HBaseTestingUtility;
+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 HBaseTestingUtility UTIL = new HBaseTestingUtility();
+
+ 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-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 996103116c5..407b0aa13d5 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,10 @@ public class DummyAsyncTable<C extends ScanResultConsumerBase> implements AsyncT
return null;
}
+ @Override
+ public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
+ Function<com.google.protobuf.RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
+ PartialResultCoprocessorCallback<S, R> callback) {
+ return null;
+ }
}