@Override
  public void getAvg(RpcController controller, AggregateRequest request,
    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;
      Long rowCountVal = 0L;
      Scan scan = ProtobufUtil.toScan(request.getScan());
      scanner = env.getRegion().getScanner(scan);
      byte[] colFamily = scan.getFamilies()[0];
      NavigableSet<byte[]> qualifiers = scan.getFamilyMap().get(colFamily);
      byte[] qualifier = null;
      if (qualifiers != null && !qualifiers.isEmpty()) {
        qualifier = qualifiers.pollFirst();
      }
      List<Cell> results = new ArrayList<>();
      boolean hasMoreRows = true;

      do {
        results.clear();
        if (shouldBreakForThrottling(request, scan, partialResultContext)) {
          break;
        }
        hasMoreRows = scanner.next(results);
        int listSize = results.size();
        for (int i = 0; i < listSize; i++) {
          sumVal =
            ci.add(sumVal, ci.castToReturnType(ci.getValue(colFamily, qualifier, results.get(i))));
        }
        rowCountVal++;
        postScanPartialResultUpdate(results, partialResultContext);
      } while (hasMoreRows);

      if (sumVal != null && !request.getClientSupportsPartialResult()) {
        AggregateResponse.Builder pair = AggregateResponse.newBuilder();
        ByteString first = ci.getProtoForPromotedType(sumVal).toByteString();
        pair.addFirstPart(first);
        ByteBuffer bb = ByteBuffer.allocate(8).putLong(rowCountVal);
        bb.rewind();
        pair.setSecondPart(ByteString.copyFrom(bb));
        response = pair.build();
      } else if (request.getClientSupportsPartialResult()) {
        AggregateResponse.Builder pair =
          responseBuilder(request, hasMoreRows, partialResultContext);
        if (sumVal != null) {
          ByteString first = ci.getProtoForPromotedType(sumVal).toByteString();
          pair.addFirstPart(first);
          ByteBuffer bb = ByteBuffer.allocate(8).putLong(rowCountVal);
          bb.rewind();
          pair.setSecondPart(ByteString.copyFrom(bb));
        }
        response = pair.build();
      }
    } catch (IOException e) {
      CoprocessorRpcUtils.setControllerException(controller, e);
    } finally {
      if (scanner != null) {
        IOUtils.closeQuietly(scanner);
      }
      closeQuota(partialResultContext);
    }
    done.run(response);
  }
Gives a Pair with first object as Sum and second object as row count, computed for a given
combination of column qualifier and column family in the given row range as defined in the Scan
object. In its current implementation, it takes one column family and one column qualifier (if
provided). In case of null column qualifier, an aggregate sum over all the entire column family
will be returned.
<p>
The average is computed in AggregationClient#avg(byte[], ColumnInterpreter, Scan) by processing
results from all regions, so its "ok" to pass sum and a Long type.
hbase-endpoint/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java