Java 类org.apache.hadoop.hbase.client.Scan 实例源码

protected void wipeOutMeta() throws IOException {
  // Mess it up by blowing up meta.
  Admin admin = TEST_UTIL.getHBaseAdmin();
  Scan s = new Scan();
  Table meta = new HTable(conf, TableName.META_TABLE_NAME);
  ResultScanner scanner = meta.getScanner(s);
  List<Delete> dels = new ArrayList<Delete>();
  for (Result r : scanner) {
    HRegionInfo info =
        HRegionInfo.getHRegionInfo(r);
    if(info != null && !info.getTable().getNamespaceAsString()
        .equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
      Delete d = new Delete(r.getRow());
      dels.add(d);
      admin.unassign(r.getRow(), true);
    }
  }
  meta.delete(dels);
  scanner.close();
  meta.close();
}

@Test(timeout = 60 * 1000)
public void testVisibilityLabelsOnRSRestart() throws Exception {
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  List<RegionServerThread> regionServerThreads = TEST_UTIL.getHBaseCluster()
      .getRegionServerThreads();
  for (RegionServerThread rsThread : regionServerThreads) {
    rsThread.getRegionServer().abort("Aborting ");
  }
  // Start one new RS
  RegionServerThread rs = TEST_UTIL.getHBaseCluster().startRegionServer();
  waitForLabelsRegionAvailability(rs.getRegionServer());
  try (Table table = createTableAndWriteDataWithLabels(tableName, "(" + SECRET + "|" + CONFIDENTIAL
      + ")", PRIVATE);) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
  }
}

/**
 * Checks that all columns have the expected value and that there is the
 * expected number of rows.
 * @throws IOException
 */
void assertExpectedTable(final Connection connection, TableName table, int count, int value)
throws IOException {
  HTableDescriptor [] htds = util.getHBaseAdmin().listTables(table.getNameAsString());
  assertEquals(htds.length, 1);
  Table t = null;
  try {
    t = connection.getTable(table);
    Scan s = new Scan();
    ResultScanner sr = t.getScanner(s);
    int i = 0;
    for (Result r : sr) {
      i++;
      for (NavigableMap<byte[], byte[]> nm : r.getNoVersionMap().values()) {
        for (byte[] val : nm.values()) {
          assertTrue(Bytes.equals(val, value(value)));
        }
      }
    }
    assertEquals(count, i);
  } catch (IOException e) {
    fail("Failed due to exception");
  } finally {
    if (t != null) t.close();
  }
}

@Test(timeout = 300000)
public void testSumWithInvalidRange() {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addFamily(TEST_FAMILY);
  scan.setStartRow(ROWS[6]);
  scan.setStopRow(ROWS[2]);
  final ColumnInterpreter<Double, Double, EmptyMsg, DoubleMsg, DoubleMsg> ci = 
      new DoubleColumnInterpreter();
  Double sum = null;
  try {
    sum = aClient.sum(TEST_TABLE, ci, scan);
  } catch (Throwable e) {
  }
  assertEquals(null, sum);// control should go to the catch block
}

/**
 * create scan on IFile
 */
private Scan createScanOnIFile(Scan rawScan, ScanRange primaryRange,
    ScanRange.ScanRangeList allRangeList) throws IOException {
  ScanRange.ScanRangeList copy = new ScanRange.ScanRangeList();
  for (ScanRange r : allRangeList.getRanges()) {
    if (r != primaryRange) {
      copy.addScanRange(r);
    }
  }
  Scan scan = new Scan();
  scan.setStartRow(primaryRange.getStart());
  scan.setStopRow(primaryRange.getStop());
  if (!copy.getRanges().isEmpty()) {
    scan.setFilter(copy.toFilterList());
  }
  scan.setCacheBlocks(false);
  return scan;
}

@Test
public void testMultiRowRangeFilterWithoutRangeOverlap() throws IOException {
  tableName = Bytes.toBytes("testMultiRowRangeFilterWithoutRangeOverlap");
  HTable ht = TEST_UTIL.createTable(tableName, family, Integer.MAX_VALUE);
  generateRows(numRows, ht, family, qf, value);

  Scan scan = new Scan();
  scan.setMaxVersions();

  List<RowRange> ranges = new ArrayList<RowRange>();
  ranges.add(new RowRange(Bytes.toBytes(30), true, Bytes.toBytes(40), false));
  ranges.add(new RowRange(Bytes.toBytes(10), true, Bytes.toBytes(20), false));
  ranges.add(new RowRange(Bytes.toBytes(60), true, Bytes.toBytes(70), false));

  MultiRowRangeFilter filter = new MultiRowRangeFilter(ranges);
  scan.setFilter(filter);
  int resultsSize = getResultsSize(ht, scan);
  LOG.info("found " + resultsSize + " results");
  List<Cell> results1 = getScanResult(Bytes.toBytes(10), Bytes.toBytes(20), ht);
  List<Cell> results2 = getScanResult(Bytes.toBytes(30), Bytes.toBytes(40), ht);
  List<Cell> results3 = getScanResult(Bytes.toBytes(60), Bytes.toBytes(70), ht);

  assertEquals(results1.size() + results2.size() + results3.size(), resultsSize);

  ht.close();
}

private int countRows(final HRegion r) throws IOException {
  int rowcount = 0;
  InternalScanner scanner = r.getScanner(new Scan());
  try {
    List<Cell> kvs = new ArrayList<Cell>();
    boolean hasNext = true;
    while (hasNext) {
      hasNext = scanner.next(kvs);
      if (!kvs.isEmpty())
        rowcount++;
    }
  } finally {
    scanner.close();
  }
  return rowcount;
}

public Result getRowOrBefore(Table table, byte[] row, byte[] family) throws IOException {
  long start = System.currentTimeMillis();
  Scan scan = new Scan();
  scan.addFamily(family);
  scan.setReversed(true);
  scan.setStartRow(row);
  scan.setCacheBlocks(false);
  scan.setCaching(1);
  scan.setSmall(true);
  ResultScanner scanner = table.getScanner(scan);
  Result ret = scanner.next();
  scanner.close();
  prevRowTotalTime += System.currentTimeMillis() - start;
  prevRowTotalCount++;
  return ret;
}

public void testSkipColumn() throws IOException {
  List<KeyValueScanner> scanners = scanFixture(kvs);
  StoreScanner scan = new StoreScanner(new Scan(), scanInfo, scanType,
      getCols("a", "d"), scanners);

  List<Cell> results = new ArrayList<Cell>();
  assertEquals(true, scan.next(results));
  assertEquals(2, results.size());
  assertEquals(kvs[0], results.get(0));
  assertEquals(kvs[3], results.get(1));
  results.clear();

  assertEquals(true, scan.next(results));
  assertEquals(1, results.size());
  assertEquals(kvs[kvs.length-1], results.get(0));

  results.clear();
  assertEquals(false, scan.next(results));
}

@Test (timeout=300000)
public void testStdWithInvalidRange() {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addFamily(TEST_FAMILY);
  scan.setStartRow(ROWS[6]);
  scan.setStopRow(ROWS[1]);
  final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci =
      new LongColumnInterpreter();
  Double std = null;
  try {
    std = aClient.std(TEST_TABLE, ci, scan);
  } catch (Throwable e) {
  }
  assertEquals(null, std);// control should go to the catch block
}

@Test
public void testGetScanner_WithNotOkFamilies() throws IOException {
  byte[] fam1 = Bytes.toBytes("fam1");
  byte[] fam2 = Bytes.toBytes("fam2");

  byte[][] families = { fam1 };

  // Setting up region
  String method = this.getName();
  this.region = initHRegion(tableName, method, CONF, families);
  try {
    Scan scan = new Scan();
    scan.addFamily(fam2);
    boolean ok = false;
    try {
      region.getScanner(scan);
    } catch (Exception e) {
      ok = true;
    }
    assertTrue("Families could not be found in Region", ok);
  } finally {
    HRegion.closeHRegion(this.region);
    this.region = null;
  }
}

@Test
public void testScanForSuperUserWithFewerLabelAuths() throws Throwable {
  String[] auths = { SECRET };
  String user = "admin";
  try (Connection conn = ConnectionFactory.createConnection(conf)) {
    VisibilityClient.setAuths(conn, auths, user);
  }
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  final Table table = createTableAndWriteDataWithLabels(tableName, SECRET + "&" + CONFIDENTIAL
      + "&!" + PRIVATE, SECRET + "&!" + PRIVATE);
  PrivilegedExceptionAction<Void> scanAction = new PrivilegedExceptionAction<Void>() {
    public Void run() throws Exception {
      Scan s = new Scan();
      s.setAuthorizations(new Authorizations(SECRET, CONFIDENTIAL));
      try (Connection connection = ConnectionFactory.createConnection(conf);
           Table t = connection.getTable(table.getName())) {
        ResultScanner scanner = t.getScanner(s);
        Result[] result = scanner.next(5);
        assertTrue(result.length == 2);
      }
      return null;
    }
  };
  SUPERUSER.runAs(scanAction);
}

private int countDeleteMarkers(Region region) throws IOException {
  Scan s = new Scan();
  s.setRaw(true);
  // use max versions from the store(s)
  s.setMaxVersions(region.getStores().iterator().next().getScanInfo().getMaxVersions());
  InternalScanner scan = region.getScanner(s);
  List<Cell> kvs = new ArrayList<Cell>();
  int res = 0;
  boolean hasMore;
  do {
    hasMore = scan.next(kvs);
    for (Cell kv : kvs) {
      if(CellUtil.isDelete(kv)) res++;
    }
    kvs.clear();
  } while (hasMore);
  scan.close();
  return res;
}

public ScanTask(ScanTaskConfig scanTaskConfig, Scan... scans) {
    if (scanTaskConfig == null) {
        throw new NullPointerException("scanTaskConfig must not be null");
    }
    if (scans == null) {
        throw new NullPointerException("scans must not be null");
    }
    if (scans.length == 0) {
        throw new IllegalArgumentException("scans must not be empty");
    }
    this.tableName = scanTaskConfig.getTableName();
    this.tableFactory = scanTaskConfig.getTableFactory();
    this.rowKeyDistributor = scanTaskConfig.getRowKeyDistributor();
    this.scans = scans;
    this.resultQueue = new ArrayBlockingQueue<>(scanTaskConfig.getScanTaskQueueSize());
}

/**
 * This will test the row count with startrow = endrow and they will be
 * non-null. The result should be 0, as it assumes a non-get query.
 * @throws Throwable
 */
@Test (timeout=300000)
public void testRowCountWithInvalidRange2() {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.setStartRow(ROWS[5]);
  scan.setStopRow(ROWS[5]);

  final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci =
      new LongColumnInterpreter();
  long rowCount = -1;
  try {
    rowCount = aClient.rowCount(TEST_TABLE, ci, scan);
  } catch (Throwable e) {
    rowCount = 0;
  }
  assertEquals(0, rowCount);
}

private void putDataAndVerify(Table table, String row, byte[] family,
    String value, int verifyNum) throws IOException {
  System.out.println("=========Putting data :" + row);
  Put put = new Put(Bytes.toBytes(row));
  put.add(family, Bytes.toBytes("q1"), Bytes.toBytes(value));
  table.put(put);
  ResultScanner resultScanner = table.getScanner(new Scan());
  List<Result> results = new ArrayList<Result>();
  while (true) {
    Result r = resultScanner.next();
    if (r == null)
      break;
    results.add(r);
  }
  resultScanner.close();
  if (results.size() != verifyNum) {
    System.out.println(results);
  }
  assertEquals(verifyNum, results.size());
}

private ResultScanner getNextScanner() throws IOException {
  if (INIT_REGION_SIZE != getRegionNumber()) {
    throw new IOException(
        "region number changed from " + INIT_REGION_SIZE + " to " + getRegionNumber());
  }
  if (regionLocationQueue.isEmpty()) return null;
  HRegionLocation regionLocation = regionLocationQueue.poll();

  Scan newScan = new Scan(rawScan);
  byte[] key = regionLocation.getRegionInfo().getStartKey();
  if (key != null && key.length > 0) newScan.setStartRow(key);
  key = regionLocation.getRegionInfo().getEndKey();
  if (key != null && key.length > 0) newScan.setStopRow(key);
  newScan.setAttribute(IndexConstants.SCAN_WITH_INDEX, Bytes.toBytes("Hi"));
  newScan.setId(rawScan.getId());
  newScan.setCacheBlocks(rawScan.getCacheBlocks());
  newScan.setCaching(rawScan.getCaching());
  return table.getScanner(newScan);
}

@Override
public InternalScanner preCompactScannerOpen(
    final ObserverContext<RegionCoprocessorEnvironment> c,
    Store store, List<? extends KeyValueScanner> scanners, ScanType scanType,
    long earliestPutTs, InternalScanner s) throws IOException {
  Long newTtl = ttls.get(store.getTableName());
  Integer newVersions = versions.get(store.getTableName());
  ScanInfo oldSI = store.getScanInfo();
  HColumnDescriptor family = store.getFamily();
  ScanInfo scanInfo = new ScanInfo(TEST_UTIL.getConfiguration(),
      family.getName(), family.getMinVersions(),
      newVersions == null ? family.getMaxVersions() : newVersions,
      newTtl == null ? oldSI.getTtl() : newTtl, family.getKeepDeletedCells(),
      oldSI.getTimeToPurgeDeletes(), oldSI.getComparator());
  Scan scan = new Scan();
  scan.setMaxVersions(newVersions == null ? oldSI.getMaxVersions() : newVersions);
  return new StoreScanner(store, scanInfo, scan, scanners, scanType,
      store.getSmallestReadPoint(), earliestPutTs);
}

@Test (timeout=300000)
public void testMinWithInvalidRange() {
  AggregationClient aClient = new AggregationClient(conf);
  Long min = null;
  Scan scan = new Scan();
  scan.addFamily(TEST_FAMILY);
  scan.setStartRow(ROWS[4]);
  scan.setStopRow(ROWS[2]);
  final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci =
      new LongColumnInterpreter();
  try {
    min = aClient.min(TEST_TABLE, ci, scan);
  } catch (Throwable e) {
  }
  assertEquals(null, min);// control should go to the catch block
}

@Override
void testRow(final int i) throws IOException {
  Pair<byte[], byte[]> startAndStopRow = getStartAndStopRow();
  Scan scan = new Scan(startAndStopRow.getFirst(), startAndStopRow.getSecond());
  scan.addColumn(FAMILY_NAME, QUALIFIER_NAME);
  ResultScanner s = this.table.getScanner(scan);
  int count = 0;
  for (Result rr = null; (rr = s.next()) != null;) {
    count++;
  }

  if (i % 100 == 0) {
    LOG.info(String.format("Scan for key range %s - %s returned %s rows",
        Bytes.toString(startAndStopRow.getFirst()),
        Bytes.toString(startAndStopRow.getSecond()), count));
  }

  s.close();
}

@Test(timeout = 300000)
public void testMinWithValidRange2WithNoCQ() throws Throwable {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addFamily(TEST_FAMILY);
  scan.setStartRow(ROWS[6]);
  scan.setStopRow(ROWS[7]);
  final ColumnInterpreter<Double, Double, EmptyMsg, DoubleMsg, DoubleMsg> ci = 
      new DoubleColumnInterpreter();
  double min = aClient.min(TEST_TABLE, ci, scan);
  assertEquals(0.60, min, 0.001);
}

private static void processTable(final FileSystem fs, final Path p, final WALFactory walFactory,
    final Configuration c, final boolean majorCompact) throws IOException {
  HRegion region;
  FSTableDescriptors fst = new FSTableDescriptors(c);
  // Currently expects tables have one region only.
  if (FSUtils.getTableName(p).equals(TableName.META_TABLE_NAME)) {
    final WAL wal =
        walFactory.getMetaWAL(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes());
    region = HRegion.newHRegion(p, wal, fs, c, HRegionInfo.FIRST_META_REGIONINFO,
        fst.get(TableName.META_TABLE_NAME), null);
  } else {
    throw new IOException("Not a known catalog table: " + p.toString());
  }
  try {
    region.mvcc.advanceTo(region.initialize(null));
    if (majorCompact) {
      region.compact(true);
    } else {
      // Default behavior
      Scan scan = new Scan();
      // scan.addFamily(HConstants.CATALOG_FAMILY);
      RegionScanner scanner = region.getScanner(scan);
      try {
        List<Cell> kvs = new ArrayList<Cell>();
        boolean done;
        do {
          kvs.clear();
          done = scanner.next(kvs);
          if (kvs.size() > 0) LOG.info(kvs);
        } while (done);
      } finally {
        scanner.close();
      }
    }
  } finally {
    region.close();
  }
}

/**
 * @throws Throwable
 */
@Test (timeout=300000)
public void testSumWithValidRange2() throws Throwable {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addColumn(TEST_FAMILY, TEST_QUALIFIER);
  scan.setStartRow(ROWS[5]);
  scan.setStopRow(ROWS[15]);
  final ColumnInterpreter<BigDecimal, BigDecimal, EmptyMsg, BigDecimalMsg, BigDecimalMsg> ci =
    new BigDecimalColumnInterpreter();
  BigDecimal sum = aClient.sum(TEST_TABLE, ci, scan);
  assertEquals(new BigDecimal("95.00"), sum);
}

@Test
public void testMultiRowRangeWithFilterListOrOperator() throws IOException {
  tableName = Bytes.toBytes("TestMultiRowRangeFilterWithFilterListOrOperator");
  HTable ht = TEST_UTIL.createTable(tableName, family, Integer.MAX_VALUE);
  generateRows(numRows, ht, family, qf, value);

  Scan scan = new Scan();
  scan.setMaxVersions();

  List<RowRange> ranges1 = new ArrayList<RowRange>();
  ranges1.add(new RowRange(Bytes.toBytes(30), true, Bytes.toBytes(40), false));
  ranges1.add(new RowRange(Bytes.toBytes(10), true, Bytes.toBytes(20), false));
  ranges1.add(new RowRange(Bytes.toBytes(60), true, Bytes.toBytes(70), false));

  MultiRowRangeFilter filter1 = new MultiRowRangeFilter(ranges1);

  List<RowRange> ranges2 = new ArrayList<RowRange>();
  ranges2.add(new RowRange(Bytes.toBytes(20), true, Bytes.toBytes(40), false));
  ranges2.add(new RowRange(Bytes.toBytes(80), true, Bytes.toBytes(90), false));

  MultiRowRangeFilter filter2 = new MultiRowRangeFilter(ranges2);

  FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
  filterList.addFilter(filter1);
  filterList.addFilter(filter2);
  scan.setFilter(filterList);
  int resultsSize = getResultsSize(ht, scan);
  LOG.info("found " + resultsSize + " results");
  List<Cell> results1 = getScanResult(Bytes.toBytes(10), Bytes.toBytes(40), ht);
  List<Cell> results2 = getScanResult(Bytes.toBytes(60), Bytes.toBytes(70), ht);
  List<Cell> results3 = getScanResult(Bytes.toBytes(80), Bytes.toBytes(90), ht);

  assertEquals(results1.size() + results2.size() + results3.size(),resultsSize);

  ht.close();
}

/**
 * Open a QuotaRetriever with the specified filter.
 * @param conf Configuration object to use.
 * @param filter the QuotaFilter
 * @return the QuotaRetriever
 * @throws IOException if a remote or network exception occurs
 */
public static QuotaRetriever open(final Configuration conf, final QuotaFilter filter)
    throws IOException {
  Scan scan = QuotaTableUtil.makeScan(filter);
  QuotaRetriever scanner = new QuotaRetriever();
  scanner.init(conf, scan);
  return scanner;
}

@Test (timeout=300000)
public void testAvgWithValidRangeWithNoCQ() throws Throwable {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addFamily(TEST_FAMILY);
  final ColumnInterpreter<BigDecimal, BigDecimal, EmptyMsg, BigDecimalMsg, BigDecimalMsg> ci =
    new BigDecimalColumnInterpreter();
  double avg = aClient.avg(TEST_TABLE, ci, scan);
  assertEquals(10.45, avg, 0.01);
}

/**
 * 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();
}

/**
 * @param scan the Scan specification
 * @param s the scanner
 * @return the scanner instance to use
 * @exception IOException Exception
 */
public RegionScanner postScannerOpen(final Scan scan, RegionScanner s) throws IOException {
  return execOperationWithResult(s,
      coprocessors.isEmpty() ? null : new RegionOperationWithResult<RegionScanner>() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      setResult(oserver.postScannerOpen(ctx, scan, getResult()));
    }
  });
}

/**
 * Create table data and run tests on specified htable using the
 * o.a.h.hbase.mapreduce API.
 *
 * @param table
 * @throws IOException
 * @throws InterruptedException
 */
static void runTestMapreduce(Table table) throws IOException,
    InterruptedException {
  org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl trr =
      new org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl();
  Scan s = new Scan();
  s.setStartRow("aaa".getBytes());
  s.setStopRow("zzz".getBytes());
  s.addFamily(FAMILY);
  trr.setScan(s);
  trr.setHTable(table);

  trr.initialize(null, null);
  Result r = new Result();
  ImmutableBytesWritable key = new ImmutableBytesWritable();

  boolean more = trr.nextKeyValue();
  assertTrue(more);
  key = trr.getCurrentKey();
  r = trr.getCurrentValue();
  checkResult(r, key, "aaa".getBytes(), "value aaa".getBytes());

  more = trr.nextKeyValue();
  assertTrue(more);
  key = trr.getCurrentKey();
  r = trr.getCurrentValue();
  checkResult(r, key, "bbb".getBytes(), "value bbb".getBytes());

  // no more data
  more = trr.nextKeyValue();
  assertFalse(more);
}

/**
 * @throws Throwable
 */
@Test (timeout=300000)
public void testMinWithValidRange2() throws Throwable {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  scan.addColumn(TEST_FAMILY, TEST_QUALIFIER);
  scan.setStartRow(ROWS[5]);
  scan.setStopRow(ROWS[15]);
  final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci =
      new LongColumnInterpreter();
  long min = aClient.min(TEST_TABLE, ci, scan);
  assertEquals(5, min);
}

/**
 * Returns a count of the rows in the region where this coprocessor is loaded.
 */
@Override
public void getRowCount(RpcController controller, ExampleProtos.CountRequest request,
                        RpcCallback<ExampleProtos.CountResponse> done) {
  Scan scan = new Scan();
  scan.setFilter(new FirstKeyOnlyFilter());
  ExampleProtos.CountResponse response = null;
  InternalScanner scanner = null;
  try {
    scanner = env.getRegion().getScanner(scan);
    List<Cell> results = new ArrayList<Cell>();
    boolean hasMore = false;
    byte[] lastRow = null;
    long count = 0;
    do {
      hasMore = scanner.next(results);
      for (Cell kv : results) {
        byte[] currentRow = CellUtil.cloneRow(kv);
        if (lastRow == null || !Bytes.equals(lastRow, currentRow)) {
          lastRow = currentRow;
          count++;
        }
      }
      results.clear();
    } while (hasMore);

    response = ExampleProtos.CountResponse.newBuilder()
        .setCount(count).build();
  } catch (IOException ioe) {
    ResponseConverter.setControllerException(controller, ioe);
  } finally {
    if (scanner != null) {
      try {
        scanner.close();
      } catch (IOException ignored) {}
    }
  }
  done.run(response);
}

/**
 * Ensure that the results returned from a scanner that retrieves all results in a single RPC call
 * matches the results that are returned from a scanner that must incrementally combine partial
 * results into complete results. A variety of scan configurations can be tested
 * @throws Exception
 */
@Test
public void testEquivalenceOfScanResults() throws Exception {
  Scan oneShotScan = new Scan();
  oneShotScan.setMaxResultSize(Long.MAX_VALUE);

  Scan partialScan = new Scan(oneShotScan);
  partialScan.setMaxResultSize(1);

  testEquivalenceOfScanResults(TABLE, oneShotScan, partialScan);
}

@Override
public KeyValueScanner preStoreScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
    Store store, final Scan scan, final NavigableSet<byte[]> targetCols, KeyValueScanner s)
    throws IOException {
  Region r = c.getEnvironment().getRegion();
  return scan.isReversed() ? new ReversedStoreScanner(store,
      store.getScanInfo(), scan, targetCols, r.getReadpoint(scan
          .getIsolationLevel())) : new StoreScanner(store,
      store.getScanInfo(), scan, targetCols, r.getReadpoint(scan
          .getIsolationLevel()));
}

@Test (timeout=300000)
public void testAvgWithValidRangeWithNullCF() {
  AggregationClient aClient = new AggregationClient(conf);
  Scan scan = new Scan();
  final ColumnInterpreter<BigDecimal, BigDecimal, EmptyMsg, BigDecimalMsg, BigDecimalMsg> ci =
    new BigDecimalColumnInterpreter();
  Double avg = null;
  try {
    avg = aClient.avg(TEST_TABLE, ci, scan);
  } catch (Throwable e) {
  }
  assertEquals(null, avg);// CP will throw an IOException about the
  // null column family, and max will be set to 0
}

private void _testBlocksScanned(HTableDescriptor table) throws Exception {
  Region r = createNewHRegion(table, START_KEY, END_KEY, TEST_UTIL.getConfiguration());
  addContent(r, FAMILY, COL);
  r.flush(true);

  CacheStats stats = new CacheConfig(TEST_UTIL.getConfiguration()).getBlockCache().getStats();
  long before = stats.getHitCount() + stats.getMissCount();
  // Do simple test of getting one row only first.
  Scan scan = new Scan(Bytes.toBytes("aaa"), Bytes.toBytes("aaz"));
  scan.addColumn(FAMILY, COL);
  scan.setMaxVersions(1);

  InternalScanner s = r.getScanner(scan);
  List<Cell> results = new ArrayList<Cell>();
  while (s.next(results))
    ;
  s.close();

  int expectResultSize = 'z' - 'a';
  assertEquals(expectResultSize, results.size());

  int kvPerBlock = (int) Math.ceil(BLOCK_SIZE / 
      (double) KeyValueUtil.ensureKeyValue(results.get(0)).getLength());
  Assert.assertEquals(2, kvPerBlock);

  long expectDataBlockRead = (long) Math.ceil(expectResultSize / (double) kvPerBlock);
  long expectIndexBlockRead = expectDataBlockRead;

  assertEquals(expectIndexBlockRead+expectDataBlockRead, stats.getHitCount() + stats.getMissCount() - before);
}

/**
 * Create a protocol buffer ScanRequest for a client Scan
 *
 * @param regionName
 * @param scan
 * @param numberOfRows
 * @param closeScanner
 * @return a scan request
 * @throws IOException
 */
public static ScanRequest buildScanRequest(final byte[] regionName, final Scan scan,
    final int numberOfRows, final boolean closeScanner) throws IOException {
  ScanRequest.Builder builder = ScanRequest.newBuilder();
  RegionSpecifier region = buildRegionSpecifier(
    RegionSpecifierType.REGION_NAME, regionName);
  builder.setNumberOfRows(numberOfRows);
  builder.setCloseScanner(closeScanner);
  builder.setRegion(region);
  builder.setScan(ProtobufUtil.toScan(scan));
  builder.setClientHandlesPartials(true);
  builder.setClientHandlesHeartbeats(true);
  builder.setTrackScanMetrics(scan.isScanMetricsEnabled());
  return builder.build();
}

/**
 * init selected range and scanner
 *
 * @throws IOException
 */
private void initScanner() throws IOException {
  ScanRange selectedRange = null;
  int selectedRegionNumber = Integer.MAX_VALUE;
  for (ScanRange range : rangeList.getRanges()) {
    int cover = countCoveringRegions(conn,
        relation.getIndexTableName(range.getFamily(), range.getQualifier()), range.getStart(),
        range.getStop());
    LOG.info("LCDBG, " + cover + " regions are covered by range " + range);
    if (selectedRegionNumber > cover) {
      selectedRegionNumber = cover;
      selectedRange = range;
    }
  }
  LOG.info("LCDBG, GC Scanner using range " + selectedRange + " with " + selectedRegionNumber
      + " regions for scan id= " + rawScan.getId());
  indexFamily = selectedRange.getFamily();
  indexQualifier = selectedRange.getQualifier();
  List<ScanRange> list = new ArrayList<>(rangeList.getRanges());
  list.remove(selectedRange);
  Scan scan = new Scan();
  scan.setStartRow(selectedRange.getStart());
  scan.setStopRow(selectedRange.getStop());
  scan.setFamilyMap(rawScan.getFamilyMap());
  scan.setCaching(rawScan.getCaching());
  scan.setCacheBlocks(rawScan.getCacheBlocks());
  scan.setId(rawScan.getId());
  scan.setFilter(new ScanRange.ScanRangeList(list).toFilterList());
  Table table = conn.getTable(
      relation.getIndexTableName(selectedRange.getFamily(), selectedRange.getQualifier()));
  scanner = table.getScanner(scan);
}

@Override
public <T> T find(TableName tableName, String family, String qualifier, final
ResultsExtractor<T> action) {
    Scan scan = new Scan();
    scan.addColumn(family.getBytes(getCharset()), qualifier.getBytes(getCharset()));
    return find(tableName, scan, action);
}

/**
 * Test the filter by adding all columns of family A in the scan. (OK)
 */
@Test
public void scanWithAllQualifiersOfFamiliyA() throws IOException {
  /* Given */
  Scan scan = new Scan();
  scan.addFamily(FAMILY_A);
  scan.setFilter(scanFilter);

  verify(scan);
}

@Override
public RegionScanner postScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
    final Scan scan, final RegionScanner s)
    throws IOException {
  ctPostScannerOpen.incrementAndGet();
  return s;
}