import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the maximum value of a column for a given column family for the
 * given range. In case qualifier is null, a max of all values for the given
 * family is returned.
 * @param table
 * @param ci
 * @param scan
 * @return max val <>
 * @throws Throwable
 *           The caller is supposed to handle the exception as they are thrown
 *           & propagated to it.
 */
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);
  class MaxCallBack implements Batch.Callback<R> {
    R max = null;

    R getMax() {
      return max;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, R result) {
      max = (max == null || (result != null && ci.compare(max, result) < 0)) ? result : max;
    }
  }
  MaxCallBack aMaxCallBack = new MaxCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, R>() {
        @Override
        public R call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getMax(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          if (response.getFirstPartCount() > 0) {
            ByteString b = response.getFirstPart(0);
            Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
            return ci.getCellValueFromProto(q);
          }
          return null;
        }
      }, aMaxCallBack);
  return aMaxCallBack.getMax();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the row count, by summing up the individual results obtained from
 * regions. In case the qualifier is null, FirstKeyValueFilter is used to
 * optimised the operation. In case qualifier is provided, I can't use the
 * filter as it may set the flag to skip to next row, but the value read is
 * not of the given filter: in this case, this particular row will not be
 * counted ==> an error.
 * @param table
 * @param ci
 * @param scan
 * @return <R, S>
 * @throws Throwable
 */
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);
  class RowNumCallback implements Batch.Callback<Long> {
    private final AtomicLong rowCountL = new AtomicLong(0);

    public long getRowNumCount() {
      return rowCountL.get();
    }

    @Override
    public void update(byte[] region, byte[] row, Long result) {
      rowCountL.addAndGet(result.longValue());
    }
  }
  RowNumCallback rowNum = new RowNumCallback();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, Long>() {
        @Override
        public Long call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getRowNum(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          byte[] bytes = getBytesFromResponse(response.getFirstPart(0));
          ByteBuffer bb = ByteBuffer.allocate(8).put(bytes);
          bb.rewind();
          return bb.getLong();
        }
      }, rowNum);
  return rowNum.getRowNumCount();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the maximum value of a column for a given column family for the
 * given range. In case qualifier is null, a max of all values for the given
 * family is returned.
 *
 * @param table
 * @param ci
 * @param scan
 * @return max val <R>
 * @throws Throwable The caller is supposed to handle the exception as they are thrown
 *                   & propagated to it.
 */
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);
    class MaxCallBack implements Batch.Callback<R> {
        R max = null;

        R getMax() {
            return max;
        }

        @Override
        public synchronized void update(byte[] region, byte[] row, R result) {
            max = (max == null || (result != null && ci.compare(max, result) < 0)) ? result : max;
        }
    }
    MaxCallBack aMaxCallBack = new MaxCallBack();
    table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
            new Batch.Call<AggregateService, R>() {
                @Override
                public R call(AggregateService instance) throws IOException {
                    ServerRpcController controller = new ServerRpcController();
                    BlockingRpcCallback<AggregateResponse> rpcCallback =
                            new BlockingRpcCallback<AggregateResponse>();
                    instance.getMax(controller, requestArg, rpcCallback);
                    AggregateResponse response = rpcCallback.get();
                    if (controller.failedOnException()) {
                        throw controller.getFailedOn();
                    }
                    if (response.getFirstPartCount() > 0) {
                        ByteString b = response.getFirstPart(0);
                        Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
                        return ci.getCellValueFromProto(q);
                    }
                    return null;
                }
            }, aMaxCallBack);
    return aMaxCallBack.getMax();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the row count, by summing up the individual results obtained from
 * regions. In case the qualifier is null, FirstKeyValueFilter is used to
 * optimised the operation. In case qualifier is provided, I can't use the
 * filter as it may set the flag to skip to next row, but the value read is
 * not of the given filter: in this case, this particular row will not be
 * counted ==> an error.
 *
 * @param table
 * @param ci
 * @param scan
 * @return <R, S>
 * @throws Throwable
 */
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);
    class RowNumCallback implements Batch.Callback<Long> {
        private final AtomicLong rowCountL = new AtomicLong(0);

        public long getRowNumCount() {
            return rowCountL.get();
        }

        @Override
        public void update(byte[] region, byte[] row, Long result) {
            rowCountL.addAndGet(result.longValue());
        }
    }
    RowNumCallback rowNum = new RowNumCallback();
    table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
            new Batch.Call<AggregateService, Long>() {
                @Override
                public Long call(AggregateService instance) throws IOException {
                    ServerRpcController controller = new ServerRpcController();
                    BlockingRpcCallback<AggregateResponse> rpcCallback =
                            new BlockingRpcCallback<AggregateResponse>();
                    instance.getRowNum(controller, requestArg, rpcCallback);
                    AggregateResponse response = rpcCallback.get();
                    if (controller.failedOnException()) {
                        throw controller.getFailedOn();
                    }
                    byte[] bytes = getBytesFromResponse(response.getFirstPart(0));
                    ByteBuffer bb = ByteBuffer.allocate(8).put(bytes);
                    bb.rewind();
                    return bb.getLong();
                }
            }, rowNum);
    return rowNum.getRowNumCount();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the maximum value of a column for a given column family for the
 * given range. In case qualifier is null, a max of all values for the given
 * family is returned.
 * @param table
 * @param ci
 * @param scan
 * @return max val <R>
 * @throws Throwable
 *           The caller is supposed to handle the exception as they are thrown
 *           & propagated to it.
 */
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);
  class MaxCallBack implements Batch.Callback<R> {
    R max = null;

    R getMax() {
      return max;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, R result) {
      max = (max == null || (result != null && ci.compare(max, result) < 0)) ? result : max;
    }
  }
  MaxCallBack aMaxCallBack = new MaxCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, R>() {
        @Override
        public R call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getMax(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          if (response.getFirstPartCount() > 0) {
            ByteString b = response.getFirstPart(0);
            Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
            return ci.getCellValueFromProto(q);
          }
          return null;
        }
      }, aMaxCallBack);
  return aMaxCallBack.getMax();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the row count, by summing up the individual results obtained from
 * regions. In case the qualifier is null, FirstKeyValueFilter is used to
 * optimised the operation. In case qualifier is provided, I can't use the
 * filter as it may set the flag to skip to next row, but the value read is
 * not of the given filter: in this case, this particular row will not be
 * counted ==> an error.
 * @param table
 * @param ci
 * @param scan
 * @return <R, S>
 * @throws Throwable
 */
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);
  class RowNumCallback implements Batch.Callback<Long> {
    private final AtomicLong rowCountL = new AtomicLong(0);

    public long getRowNumCount() {
      return rowCountL.get();
    }

    @Override
    public void update(byte[] region, byte[] row, Long result) {
      rowCountL.addAndGet(result.longValue());
    }
  }
  RowNumCallback rowNum = new RowNumCallback();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, Long>() {
        @Override
        public Long call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getRowNum(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          byte[] bytes = getBytesFromResponse(response.getFirstPart(0));
          ByteBuffer bb = ByteBuffer.allocate(8).put(bytes);
          bb.rewind();
          return bb.getLong();
        }
      }, rowNum);
  return rowNum.getRowNumCount();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the maximum value of a column for a given column family for the
 * given range. In case qualifier is null, a max of all values for the given
 * family is returned.
 * @param table
 * @param ci
 * @param scan
 * @return max val <R>
 * @throws Throwable
 *           The caller is supposed to handle the exception as they are thrown
 *           & propagated to it.
 */
public <R, S, P extends Message, Q extends Message, T extends Message> 
R max(final HTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
    final Scan scan) throws Throwable {
  final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false);
  class MaxCallBack implements Batch.Callback<R> {
    R max = null;

    R getMax() {
      return max;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, R result) {
      max = (max == null || (result != null && ci.compare(max, result) < 0)) ? result : max;
    }
  }
  MaxCallBack aMaxCallBack = new MaxCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, R>() {
        @Override
        public R call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback = 
              new BlockingRpcCallback<AggregateResponse>();
          instance.getMax(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          if (response.getFirstPartCount() > 0) {
            ByteString b = response.getFirstPart(0);
            Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
            return ci.getCellValueFromProto(q);
          }
          return null;
        }
      }, aMaxCallBack);
  return aMaxCallBack.getMax();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the row count, by summing up the individual results obtained from
 * regions. In case the qualifier is null, FirstKeyValueFilter is used to
 * optimised the operation. In case qualifier is provided, I can't use the
 * filter as it may set the flag to skip to next row, but the value read is
 * not of the given filter: in this case, this particular row will not be
 * counted ==> an error.
 * @param table
 * @param ci
 * @param scan
 * @return <R, S>
 * @throws Throwable
 */
public <R, S, P extends Message, Q extends Message, T extends Message> 
long rowCount(final HTable table,
    final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
  final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, true);
  class RowNumCallback implements Batch.Callback<Long> {
    private final AtomicLong rowCountL = new AtomicLong(0);

    public long getRowNumCount() {
      return rowCountL.get();
    }

    @Override
    public void update(byte[] region, byte[] row, Long result) {
      rowCountL.addAndGet(result.longValue());
    }
  }
  RowNumCallback rowNum = new RowNumCallback();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, Long>() {
        @Override
        public Long call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback = 
              new BlockingRpcCallback<AggregateResponse>();
          instance.getRowNum(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          byte[] bytes = getBytesFromResponse(response.getFirstPart(0));
          ByteBuffer bb = ByteBuffer.allocate(8).put(bytes);
          bb.rewind();
          return bb.getLong();
        }
      }, rowNum);
  return rowNum.getRowNumCount();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the maximum value of a column for a given column family for the
 * given range. In case qualifier is null, a max of all values for the given
 * family is returned.
 * @param table
 * @param ci
 * @param scan
 * @return max val <>
 * @throws Throwable
 *           The caller is supposed to handle the exception as they are thrown
 *           & propagated to it.
 */
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);
  class MaxCallBack implements Batch.Callback<R> {
    R max = null;

    R getMax() {
      return max;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, R result) {
      max = (max == null || (result != null && ci.compare(max, result) < 0)) ? result : max;
    }
  }
  MaxCallBack aMaxCallBack = new MaxCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, R>() {
        @Override
        public R call(AggregateService instance) throws IOException {
          RpcController controller = new AggregationClientRpcController();
          CoprocessorRpcUtils.BlockingRpcCallback<AggregateResponse> rpcCallback =
              new CoprocessorRpcUtils.BlockingRpcCallback<>();
          instance.getMax(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failed()) {
            throw new IOException(controller.errorText());
          }
          if (response.getFirstPartCount() > 0) {
            ByteString b = response.getFirstPart(0);
            Q q = getParsedGenericInstance(ci.getClass(), 3, b);
            return ci.getCellValueFromProto(q);
          }
          return null;
        }
      }, aMaxCallBack);
  return aMaxCallBack.getMax();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the row count, by summing up the individual results obtained from
 * regions. In case the qualifier is null, FirstKeyValueFilter is used to
 * optimised the operation. In case qualifier is provided, I can't use the
 * filter as it may set the flag to skip to next row, but the value read is
 * not of the given filter: in this case, this particular row will not be
 * counted ==> an error.
 * @param table
 * @param ci
 * @param scan
 * @return <R, S>
 * @throws Throwable
 */
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);
  class RowNumCallback implements Batch.Callback<Long> {
    private final AtomicLong rowCountL = new AtomicLong(0);

    public long getRowNumCount() {
      return rowCountL.get();
    }

    @Override
    public void update(byte[] region, byte[] row, Long result) {
      rowCountL.addAndGet(result.longValue());
    }
  }
  RowNumCallback rowNum = new RowNumCallback();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, Long>() {
        @Override
        public Long call(AggregateService instance) throws IOException {
          RpcController controller = new AggregationClientRpcController();
          CoprocessorRpcUtils.BlockingRpcCallback<AggregateResponse> rpcCallback =
              new CoprocessorRpcUtils.BlockingRpcCallback<>();
          instance.getRowNum(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failed()) {
            throw new IOException(controller.errorText());
          }
          byte[] bytes = getBytesFromResponse(response.getFirstPart(0));
          ByteBuffer bb = ByteBuffer.allocate(8).put(bytes);
          bb.rewind();
          return bb.getLong();
        }
      }, rowNum);
  return rowNum.getRowNumCount();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
public static <R, S, P extends Message, Q extends Message, T extends Message> CompletableFuture<R>
    max(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
  CompletableFuture<R> future = new CompletableFuture<>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, false);
  } catch (IOException e) {
    future.completeExceptionally(e);
    return future;
  }
  AbstractAggregationCallback<R> callback = new AbstractAggregationCallback<R>(future) {

    private R max;

    @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 R getFinalResult() {
      return max;
    }
  };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
public static <R, S, P extends Message, Q extends Message, T extends Message> CompletableFuture<R>
    min(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
  CompletableFuture<R> future = new CompletableFuture<>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, false);
  } catch (IOException e) {
    future.completeExceptionally(e);
    return future;
  }
  AbstractAggregationCallback<R> callback = new AbstractAggregationCallback<R>(future) {

    private R min;

    @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 R getFinalResult() {
      return min;
    }
  };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
public static <R, S, P extends Message, Q extends Message, T extends Message>
    CompletableFuture<Long>
    rowCount(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
  CompletableFuture<Long> future = new CompletableFuture<>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, true);
  } catch (IOException e) {
    future.completeExceptionally(e);
    return future;
  }
  AbstractAggregationCallback<Long> callback = new AbstractAggregationCallback<Long>(future) {

    private long count;

    @Override
    protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
      count += resp.getFirstPart(0).asReadOnlyByteBuffer().getLong();
    }

    @Override
    protected Long getFinalResult() {
      return count;
    }
  };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
public static <R, S, P extends Message, Q extends Message, T extends Message> CompletableFuture<S>
    sum(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
  CompletableFuture<S> future = new CompletableFuture<>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, false);
  } catch (IOException e) {
    future.completeExceptionally(e);
    return future;
  }
  AbstractAggregationCallback<S> callback = new AbstractAggregationCallback<S>(future) {

    private S sum;

    @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 S getFinalResult() {
      return sum;
    }
  };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
public static <R, S, P extends Message, Q extends Message, T extends Message>
    CompletableFuture<Double>
    avg(AsyncTable<?> table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) {
  CompletableFuture<Double> future = new CompletableFuture<>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, false);
  } catch (IOException e) {
    future.completeExceptionally(e);
    return future;
  }
  AbstractAggregationCallback<Double> callback = new AbstractAggregationCallback<Double>(future) {

    private S sum;

    long count = 0L;

    @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 Double getFinalResult() {
      return ci.divideForAvg(sum, count);
    }
  };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
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>>();
  AggregateRequest req;
  try {
    req = validateArgAndGetPB(scan, ci, false);
  } 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) {

        private final NavigableMap<byte[], S> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);

        @Override
        protected void aggregate(RegionInfo region, AggregateResponse resp) throws IOException {
          if (resp.getFirstPartCount() > 0) {
            map.put(region.getStartKey(), getPromotedValueFromProto(ci, resp, firstPartIndex));
          }
        }

        @Override
        protected NavigableMap<byte[], S> getFinalResult() {
          return map;
        }
      };
  table
      .<AggregateService, AggregateResponse> 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();
  return future;
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It gives the minimum value of a column for a given column family for the
 * given range. In case qualifier is null, a min of all values for the given
 * family is returned.
 * @param table
 * @param ci
 * @param scan
 * @return min val <R>
 * @throws Throwable
 */
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);
  class MinCallBack implements Batch.Callback<R> {

    private R min = null;

    public R getMinimum() {
      return min;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, R result) {
      min = (min == null || (result != null && ci.compare(result, min) < 0)) ? result : min;
    }
  }
  MinCallBack minCallBack = new MinCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, R>() {

        @Override
        public R call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getMin(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          if (response.getFirstPartCount() > 0) {
            ByteString b = response.getFirstPart(0);
            Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
            return ci.getCellValueFromProto(q);
          }
          return null;
        }
      }, minCallBack);
  log.debug("Min fom all regions is: " + minCallBack.getMinimum());
  return minCallBack.getMinimum();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It sums up the value returned from various regions. In case qualifier is
 * null, summation of all the column qualifiers in the given family is done.
 * @param table
 * @param ci
 * @param scan
 * @return sum <S>
 * @throws Throwable
 */
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);

  class SumCallBack implements Batch.Callback<S> {
    S sumVal = null;

    public S getSumResult() {
      return sumVal;
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, S result) {
      sumVal = ci.add(sumVal, result);
    }
  }
  SumCallBack sumCallBack = new SumCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, S>() {
        @Override
        public S call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getSum(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          if (response.getFirstPartCount() == 0) {
            return null;
          }
          ByteString b = response.getFirstPart(0);
          T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
          S s = ci.getPromotedValueFromProto(t);
          return s;
        }
      }, sumCallBack);
  return sumCallBack.getSumResult();
}
 

import org.apache.hadoop.hbase.protobuf.generated.AggregateProtos.AggregateService; //导入依赖的package包/类
/**
 * It computes average while fetching sum and row count from all the
 * corresponding regions. Approach is to compute a global sum of region level
 * sum and rowcount and then compute the average.
 * @param table
 * @param scan
 * @throws Throwable
 */
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);
  class AvgCallBack implements Batch.Callback<Pair<S, Long>> {
    S sum = null;
    Long rowCount = 0l;

    public synchronized Pair<S, Long> getAvgArgs() {
      return new Pair<S, Long>(sum, rowCount);
    }

    @Override
    public synchronized void update(byte[] region, byte[] row, Pair<S, Long> result) {
      sum = ci.add(sum, result.getFirst());
      rowCount += result.getSecond();
    }
  }
  AvgCallBack avgCallBack = new AvgCallBack();
  table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.getStopRow(),
      new Batch.Call<AggregateService, Pair<S, Long>>() {
        @Override
        public Pair<S, Long> call(AggregateService instance) throws IOException {
          ServerRpcController controller = new ServerRpcController();
          BlockingRpcCallback<AggregateResponse> rpcCallback =
              new BlockingRpcCallback<AggregateResponse>();
          instance.getAvg(controller, requestArg, rpcCallback);
          AggregateResponse response = rpcCallback.get();
          if (controller.failedOnException()) {
            throw controller.getFailedOn();
          }
          Pair<S, Long> pair = new Pair<S, Long>(null, 0L);
          if (response.getFirstPartCount() == 0) {
            return pair;
          }
          ByteString b = response.getFirstPart(0);
          T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
          S s = ci.getPromotedValueFromProto(t);
          pair.setFirst(s);
          ByteBuffer bb = ByteBuffer.allocate(8).put(
              getBytesFromResponse(response.getSecondPart()));
          bb.rewind();
          pair.setSecond(bb.getLong());
          return pair;
        }
      }, avgCallBack);
  return avgCallBack.getAvgArgs();
}