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

private void verify(final Table table) throws IOException {
  Scan scan = new Scan();
  scan.addColumn(FAMILY_NAME, COLUMN_NAME);
  scan.setMaxVersions(1);
  ResultScanner scanner = table.getScanner(scan);
  for (Result r: scanner) {
    for (Cell kv : r.listCells()) {
      log.debug(Bytes.toString(r.getRow()) + "\t" + Bytes.toString(CellUtil.cloneFamily(kv))
          + "\t" + Bytes.toString(CellUtil.cloneQualifier(kv))
          + "\t" + kv.getTimestamp() + "\t" + Bytes.toBoolean(CellUtil.cloneValue(kv)));
      org.junit.Assert.assertEquals(TIMESTAMP.get(kv.getTimestamp()),
        (Boolean)Bytes.toBoolean(CellUtil.cloneValue(kv)));
    }
  }
  scanner.close();
}

public static void QueryAll(String tableName) {
    try {
        HTableInterface table = conn.getTable(tableName);

        ResultScanner rs = table.getScanner(new Scan());
        for (Result r : rs) {
            System.out.println("rowkey:" + new String(r.getRow()));
            for (KeyValue keyValue : r.raw()) {
                System.out.println("column：" + new String(keyValue.getFamily())
                        + "====value:" + new String(keyValue.getValue()));
            }
        }
        table.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

public static void QueryByCondition2(String tableName) {

        try {
            HTablePool pool = new HTablePool(configuration, 1000);
            HTable table = (HTable) pool.getTable(tableName);
            Filter filter = new SingleColumnValueFilter(Bytes
                    .toBytes("column1"), null, CompareOp.EQUAL, Bytes
                    .toBytes("aaa")); // 当列column1的值为aaa时进行查询
            Scan s = new Scan();
            s.setFilter(filter);
            ResultScanner rs = table.getScanner(s);
            for (Result r : rs) {
                System.out.println("获得到rowkey:" + new String(r.getRow()));
                for (KeyValue keyValue : r.raw()) {
                    System.out.println("列：" + new String(keyValue.getFamily())
                            + "====值:" + new String(keyValue.getValue()));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

private static void waitUntilAllRegionsAssigned()
throws IOException {
  HTable meta = new HTable(TEST_UTIL.getConfiguration(), TableName.META_TABLE_NAME);
  while (true) {
    int rows = 0;
    Scan scan = new Scan();
    scan.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
    ResultScanner s = meta.getScanner(scan);
    for (Result r = null; (r = s.next()) != null;) {
      byte [] b =
        r.getValue(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
      if (b == null || b.length <= 0) {
        break;
      }
      rows++;
    }
    s.close();
    // If I get to here and all rows have a Server, then all have been assigned.
    if (rows >= countOfRegions) {
      break;
    }
    LOG.info("Found=" + rows);
    Threads.sleep(1000);
  }
  meta.close();
}

public void testEquivalenceOfScanResults(Table table, Scan scan1, Scan scan2) throws Exception {
  ResultScanner scanner1 = table.getScanner(scan1);
  ResultScanner scanner2 = table.getScanner(scan2);

  Result r1 = null;
  Result r2 = null;
  int count = 0;

  while ((r1 = scanner1.next()) != null) {
    r2 = scanner2.next();

    assertTrue(r2 != null);
    compareResults(r1, r2, "Comparing result #" + count);
    count++;
  }

  r2 = scanner2.next();
  assertTrue("r2: " + r2 + " Should be null", r2 == null);

  scanner1.close();
  scanner2.close();
}

/**
 * 列族过滤器
 *
 * @param tableName 表名
 * @param rowFamily 列族
 * @param count     数量
 */
public void familyFilter(String tableName, String rowFamily, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    //使用列族过滤器
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER, new BinaryComparator(Bytes.toBytes(rowFamily))));//直接行健
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new RegexStringComparator("row.*")));//正则表达式
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new SubstringComparator("row")));//字符串包含
    scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator("mm".getBytes())));//字符串前缀
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/**
 * 列限定符过滤器
 *
 * @param tableName  表名
 * @param columnName 列限定符
 * @param count      数量
 */
public void qualifierFilter(String tableName, String columnName, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    //使用列族过滤器
    scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(columnName))));//直接行健
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("row.*")));//正则表达式
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("row")));//字符串包含
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator("m".getBytes())));//字符串前缀
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/**
 * 参考列过滤器(获取相同时间戳的列)
 *
 * @param tableName    表名
 * @param columnFamily 列族
 * @param qualifier    列限定符
 * @param columnValue  列值
 * @param count        数量
 */
public void dependentColumnFilter(String tableName, String columnFamily, String qualifier, String columnValue, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    scan.setFilter(new PrefixFilter(Bytes.toBytes("")));
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/**
 * 单列排除过滤器（返回的列 不包含参考列）
 *
 * @param tableName    表名
 * @param columnFamily 列族
 * @param qualifier    列限定符
 * @param columnValue  列值
 * @param count        数量
 */
public void SingleColumnValueExcludeFilter(String tableName, String columnFamily, String qualifier, String columnValue, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    SingleColumnValueExcludeFilter singleColumnValueFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(columnValue));
    //singleColumnValueFilter.setFilterIfMissing(true);//当不存在这列的行 默认不过滤
    singleColumnValueFilter.setLatestVersionOnly(true);//获取最新版本
    scan.setFilter(singleColumnValueFilter);
    scan.setCaching(10);
    //scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/**
 * When a scan has a filter where {@link org.apache.hadoop.hbase.filter.Filter#hasFilterRow()} is
 * true, the scanner should not return partial results. The scanner cannot return partial results
 * because the entire row needs to be read for the include/exclude decision to be made
 */
@Test
public void testNoPartialResultsWhenRowFilterPresent() throws Exception {
  Scan scan = new Scan();
  scan.setMaxResultSize(1);
  scan.setAllowPartialResults(true);
  // If a filter hasFilter() is true then partial results should not be returned else filter
  // application server side would break.
  scan.setFilter(new RandomRowFilter(1.0f));
  ResultScanner scanner = TABLE.getScanner(scan);

  Result r = null;
  while ((r = scanner.next()) != null) {
    assertFalse(r.isPartial());
  }

  scanner.close();
}

/**
 * 拷贝表
 * 
 * @throws IOException
 */
public static void copyTable(String oldTableName, String newTableName,String ColumnFamily, String ColumnName)throws IOException {
    if(CreateNewTable(newTableName))
        logger.info("创建表"+newTableName+"表成功");
    else{
        logger.info("创建表"+newTableName+"表失败");
    }
    Scan s = new Scan();
    s.addColumn(Bytes.toBytes(ColumnFamily), Bytes.toBytes(ColumnName));
    s.setMaxVersions(1);
    s.setCacheBlocks(false);
    ResultScanner rs = hbase_table.getScanner(s);

    HTableInterface hbase_table_new = conn.getTable(newTableName);
    for (Result r : rs) {
        byte[] key = r.getRow();
        byte[] value = r.getValue(Bytes.toBytes(ColumnFamily), Bytes.toBytes(ColumnName));
        Put put = new Put(key);
        put.add(Bytes.toBytes(ColumnFamily), Bytes.toBytes(ColumnName), value);
        hbase_table_new.put(put);
    }
    rs.close();
    hbase_table_new.close();
}

@Test(timeout = 300000)
public void testHBASE14489() throws IOException {
  TableName tableName = TableName.valueOf("testHBASE14489");
  HTable table = util.createTable(tableName, new byte[][] { A });
  Put put = new Put(ROW);
  put.addColumn(A, A, A);
  table.put(put);

  Scan s = new Scan();
  s.setFilter(new FilterAllFilter());
  ResultScanner scanner = table.getScanner(s);
  try {
    for (Result rr = scanner.next(); rr != null; rr = scanner.next()) {
    }
  } finally {
    scanner.close();
  }
  verifyMethodResult(SimpleRegionObserver.class, new String[] { "wasScannerFilterRowCalled" },
    tableName, new Boolean[] { true });
  util.deleteTable(tableName);
  table.close();
}

@Override
public <T> List<T> find(TableName tableName, final List<Scan> scanList, final
ResultsExtractor<T> action) {
    assertAccessAvailable();
    return execute(tableName, new TableCallback<List<T>>() {
        @Override
        public List<T> doInTable(Table table) throws Throwable {
            List<T> result = new ArrayList<>(scanList.size());
            for (Scan scan : scanList) {
                final ResultScanner scanner = table.getScanner(scan);
                try {
                    T t = action.extractData(scanner);
                    result.add(t);
                } finally {
                    scanner.close();
                }
            }
            return result;
        }
    });
}

private ResultScanner[] splitScan(Table table, Scan originalScan, AbstractRowKeyDistributor
        rowKeyDistributor) throws IOException {
    Scan[] scans = rowKeyDistributor.getDistributedScans(originalScan);
    final int length = scans.length;
    for (int i = 0; i < length; i++) {
        Scan scan = scans[i];
        // other properties are already set upon construction
        scan.setId(scan.getId() + "-" + i);
    }

    ResultScanner[] scanners = new ResultScanner[length];
    boolean success = false;
    try {
        for (int i = 0; i < length; i++) {
            scanners[i] = table.getScanner(scans[i]);
        }
        success = true;
    } finally {
        if (!success) {
            closeScanner(scanners);
        }
    }
    return scanners;
}

private void doRawScan() throws IOException {
  FilterList filterList = new FilterList();
  CompareFilter.CompareOp startOp = CompareFilter.CompareOp.GREATER_OR_EQUAL;
  CompareFilter.CompareOp stopOp = CompareFilter.CompareOp.LESS_OR_EQUAL;
  for (int i = 0; i < indexColumnNames.length && i < scanValues.length; i++) {
    filterList.addFilter(
        new SingleColumnValueFilter(familyName, Bytes.toBytes(indexColumnNames[i]), startOp,
            Bytes.toBytes(scanValues[i][0])));
    filterList.addFilter(
        new SingleColumnValueFilter(familyName, Bytes.toBytes(indexColumnNames[i]), stopOp,
            Bytes.toBytes(scanValues[i][1])));
  }
  Scan scan = new Scan();
  scan.setFilter(filterList);
  scan.setId("raw-scan");
  Table table = conn.getTable(tableName);
  ResultScanner scanner = table.getScanner(scan);
  Result result;
  int count = 0;
  while ((result = scanner.next()) != null) {
    ++count;
    if (PRINT_RESULT) printResult(result);
  }
  scanner.close();
  System.out.println("raw scan has " + count + " records");
}

private void verifyScanResult(Table table, Scan scan, List<byte[]> expectedKeys, String message) {
  List<byte[]> actualKeys = new ArrayList<>();
  try {
    ResultScanner scanner = table.getScanner(scan);
    for (Result result : scanner) {
      actualKeys.add(result.getRow());
    }

    String fullMessage = message;
    if (LOG.isDebugEnabled()) {
      fullMessage = message + "\n" + tableOfTwoListsOfByteArrays(
              "Expected", expectedKeys,
              "Actual  ", actualKeys);
    }

    Assert.assertArrayEquals(
            fullMessage,
            expectedKeys.toArray(),
            actualKeys.toArray());
  } catch (IOException e) {
    e.printStackTrace();
    Assert.fail();
  }
}

@Override
public List<TResult> getScannerResults(ByteBuffer table, TScan scan, int numRows)
    throws TIOError, TException {
  Table htable = getTable(table);
  List<TResult> results = null;
  ResultScanner scanner = null;
  try {
    scanner = htable.getScanner(scanFromThrift(scan));
    results = resultsFromHBase(scanner.next(numRows));
  } catch (IOException e) {
    throw getTIOError(e);
  } finally {
    if (scanner != null) {
      scanner.close();
    }
    closeTable(htable);
  }
  return results;
}

@Test(timeout = 60 * 1000)
public void testVisibilityLabelsOnWALReplay() throws Exception {
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName,
      "(" + SECRET + "|" + CONFIDENTIAL + ")", PRIVATE);) {
    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());
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
  }
}

private static CINode findStartNode(Table table, byte[] startKey) throws IOException {
  Scan scan = new Scan();
  scan.setStartRow(startKey);
  scan.setBatch(1);
  scan.addColumn(FAMILY_NAME, COLUMN_PREV);

  long t1 = System.currentTimeMillis();
  ResultScanner scanner = table.getScanner(scan);
  Result result = scanner.next();
  long t2 = System.currentTimeMillis();
  scanner.close();

  if ( result != null) {
    CINode node = getCINode(result, new CINode());
    System.out.printf("FSR %d %s\n", t2 - t1, Bytes.toStringBinary(node.key));
    return node;
  }

  System.out.println("FSR " + (t2 - t1));

  return null;
}

/**
 * Insert a mix of puts and deletes
 * @throws Exception
 */
@Test
public void testMixedPutDelete() throws Exception {
  List<WALEntry> entries = new ArrayList<WALEntry>(BATCH_SIZE/2);
  List<Cell> cells = new ArrayList<Cell>();
  for(int i = 0; i < BATCH_SIZE/2; i++) {
    entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
  }
  SINK.replicateEntries(entries, CellUtil.createCellScanner(cells));

  entries = new ArrayList<WALEntry>(BATCH_SIZE);
  cells = new ArrayList<Cell>();
  for(int i = 0; i < BATCH_SIZE; i++) {
    entries.add(createEntry(TABLE_NAME1, i,
        i % 2 != 0 ? KeyValue.Type.Put: KeyValue.Type.DeleteColumn, cells));
  }

  SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()));
  Scan scan = new Scan();
  ResultScanner scanRes = table1.getScanner(scan);
  assertEquals(BATCH_SIZE/2, scanRes.next(BATCH_SIZE).length);
}

public void testExpectedValuesOfPartialResults(boolean reversed) throws Exception {
  Scan partialScan = new Scan();
  partialScan.setMaxVersions();
  // Max result size of 1 ensures that each RPC request will return a single cell. The scanner
  // will need to reconstruct the results into a complete result before returning to the caller
  partialScan.setMaxResultSize(1);
  partialScan.setReversed(reversed);
  ResultScanner partialScanner = TABLE.getScanner(partialScan);

  final int startRow = reversed ? ROWS.length - 1 : 0;
  final int endRow = reversed ? -1 : ROWS.length;
  final int loopDelta = reversed ? -1 : 1;
  String message;

  for (int row = startRow; row != endRow; row = row + loopDelta) {
    message = "Ensuring the expected keyValues are present for row " + row;
    List<Cell> expectedKeyValues = createKeyValuesForRow(ROWS[row], FAMILIES, QUALIFIERS, VALUE);
    Result result = partialScanner.next();
    assertFalse(result.isPartial());
    verifyResult(result, expectedKeyValues, message);
  }

  partialScanner.close();
}

/**
 * Insert then do different types of deletes
 * @throws Exception
 */
@Test
public void testMixedDeletes() throws Exception {
  List<WALEntry> entries = new ArrayList<WALEntry>(3);
  List<Cell> cells = new ArrayList<Cell>();
  for(int i = 0; i < 3; i++) {
    entries.add(createEntry(TABLE_NAME1, i, KeyValue.Type.Put, cells));
  }
  SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()));
  entries = new ArrayList<WALEntry>(3);
  cells = new ArrayList<Cell>();
  entries.add(createEntry(TABLE_NAME1, 0, KeyValue.Type.DeleteColumn, cells));
  entries.add(createEntry(TABLE_NAME1, 1, KeyValue.Type.DeleteFamily, cells));
  entries.add(createEntry(TABLE_NAME1, 2, KeyValue.Type.DeleteColumn, cells));

  SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator()));

  Scan scan = new Scan();
  ResultScanner scanRes = table1.getScanner(scan);
  assertEquals(0, scanRes.next(3).length);
}

@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);
  }
}

/**
 * This verifies that each cell has a tag that is equal to its rowkey name.  For this to work
 * the hbase instance must have HConstants.RPC_CODEC_CONF_KEY set to
 * KeyValueCodecWithTags.class.getCanonicalName());
 * @param table table containing tagged cells
 * @throws IOException if problems reading table
 */
public static void verifyTags(Table table) throws IOException {
  ResultScanner s = table.getScanner(new Scan());
  for (Result r : s) {
    for (Cell c : r.listCells()) {
      byte[] ta = c.getTagsArray();
      int toff = c.getTagsOffset();
      int tlen = c.getTagsLength();
      Tag t = Tag.getTag(ta, toff, tlen, TagType.ACL_TAG_TYPE);
      if (t == null) {
        fail(c.toString() + " has null tag");
        continue;
      }
      byte[] tval = t.getValue();
      assertArrayEquals(c.toString() + " has tag" + Bytes.toString(tval),
          r.getRow(), tval);
    }
  }
}

/**
 * Run the scan to completetion and check the metric against the specified value
 * @param scan
 * @param metricKey
 * @param expectedValue
 * @throws Exception
 */
public void testMetric(Scan scan, String metricKey, long expectedValue) throws Exception {
  assertTrue("Scan should be configured to record metrics", scan.isScanMetricsEnabled());
  ResultScanner scanner = TABLE.getScanner(scan);

  // Iterate through all the results
  for (Result r : scanner) {
  }
  scanner.close();
  ScanMetrics metrics = scan.getScanMetrics();
  assertTrue("Metrics are null", metrics != null);
  assertTrue("Metric : " + metricKey + " does not exist", metrics.hasCounter(metricKey));
  final long actualMetricValue = metrics.getCounter(metricKey).get();
  assertEquals("Metric: " + metricKey + " Expected: " + expectedValue + " Actual: "
      + actualMetricValue, expectedValue, actualMetricValue);

}

@Test
public void testSimpleVisibilityLabels() throws Exception {
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName, SECRET + "|" + CONFIDENTIAL,
      PRIVATE + "|" + CONFIDENTIAL)) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, CONFIDENTIAL, PRIVATE));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);

    assertTrue(next.length == 2);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row1, 0, row1.length));
    cellScanner = next[1].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
  }
}

public void testPartialResultsAndBatch(final int batch, final int cellsPerPartialResult)
    throws Exception {
  if (LOG.isInfoEnabled()) {
    LOG.info("batch: " + batch + " cellsPerPartialResult: " + cellsPerPartialResult);
  }

  Scan scan = new Scan();
  scan.setMaxResultSize(getResultSizeForNumberOfCells(cellsPerPartialResult));
  scan.setBatch(batch);
  ResultScanner scanner = TABLE.getScanner(scan);
  Result result = scanner.next();
  int repCount = 0;

  while ((result = scanner.next()) != null) {
    assertTrue(result.rawCells() != null);

    if (result.isPartial()) {
      final String error =
          "Cells:" + result.rawCells().length + " Batch size:" + batch
              + " cellsPerPartialResult:" + cellsPerPartialResult + " rep:" + repCount;
      assertTrue(error, result.rawCells().length <= Math.min(batch, cellsPerPartialResult));
    } else {
      assertTrue(result.rawCells().length <= batch);
    }
    repCount++;
  }

  scanner.close();
}

/**
 * Performs a full scan of a catalog table.
 * @param connection connection we're using
 * @param visitor Visitor invoked against each row.
 * @param startrow Where to start the scan. Pass null if want to begin scan
 * at first row.
 * <code>hbase:meta</code>, the default (pass false to scan hbase:meta)
 * @throws IOException
 */
public static void fullScan(Connection connection,
  final Visitor visitor, final byte [] startrow)
throws IOException {
  Scan scan = new Scan();
  if (startrow != null) scan.setStartRow(startrow);
  if (startrow == null) {
    int caching = connection.getConfiguration()
        .getInt(HConstants.HBASE_META_SCANNER_CACHING, 100);
    scan.setCaching(caching);
  }
  scan.addFamily(HConstants.CATALOG_FAMILY);
  Table metaTable = getMetaHTable(connection);
  ResultScanner scanner = null;
  try {
    scanner = metaTable.getScanner(scan);
    Result data;
    while((data = scanner.next()) != null) {
      if (data.isEmpty()) continue;
      // Break if visit returns false.
      if (!visitor.visit(data)) break;
    }
  } finally {
    if (scanner != null) scanner.close();
    metaTable.close();
  }
}

/**
 * Create a table that IOE's on first scanner next call
 *
 * @throws IOException
 */
static Table createIOEScannerTable(byte[] name, final int failCnt)
    throws IOException {
  // build up a mock scanner stuff to fail the first time
  Answer<ResultScanner> a = new Answer<ResultScanner>() {
    int cnt = 0;

    @Override
    public ResultScanner answer(InvocationOnMock invocation) throws Throwable {
      // first invocation return the busted mock scanner
      if (cnt++ < failCnt) {
        // create mock ResultScanner that always fails.
        Scan scan = mock(Scan.class);
        doReturn("bogus".getBytes()).when(scan).getStartRow(); // avoid npe
        ResultScanner scanner = mock(ResultScanner.class);
        // simulate TimeoutException / IOException
        doThrow(new IOException("Injected exception")).when(scanner).next();
        return scanner;
      }

      // otherwise return the real scanner.
      return (ResultScanner) invocation.callRealMethod();
    }
  };

  Table htable = spy(createTable(name));
  doAnswer(a).when(htable).getScanner((Scan) anyObject());
  return htable;
}

private void deleteAllRows(final Table table) throws IOException {
    final Scan scan = new Scan();
    final List<Delete> deleteList = new ArrayList<>();
    final ResultScanner results = table.getScanner(scan);
    for (final Result result : results) {
        deleteList.add(new Delete(result.getRow()));
    }
    results.close();
    table.delete(deleteList);
}

private int countRows(final Table table) throws IOException {
    int count = 0;
    final Scan scan = new Scan();
    scan.addFamily(EventStoreColumnFamily.COUNTS.asByteArray());
    scan.addFamily(EventStoreColumnFamily.VALUES.asByteArray());
    try (final ResultScanner results = table.getScanner(scan)) {
        count = Iterables.size(results);
    }
    return count;
}

/**
 * @param resultSizeRowLimit The row limit that will be enforced through maxResultSize
 * @param cachingRowLimit The row limit that will be enforced through caching
 * @throws Exception
 */
public void testPartialResultsAndCaching(int resultSizeRowLimit, int cachingRowLimit)
    throws Exception {
  Scan scan = new Scan();
  scan.setAllowPartialResults(true);

  // The number of cells specified in the call to getResultSizeForNumberOfCells is offset to
  // ensure that the result size we specify is not an exact multiple of the number of cells
  // in a row. This ensures that partial results will be returned when the result size limit
  // is reached before the caching limit.
  int cellOffset = NUM_COLS / 3;
  long maxResultSize = getResultSizeForNumberOfCells(resultSizeRowLimit * NUM_COLS + cellOffset);
  scan.setMaxResultSize(maxResultSize);
  scan.setCaching(cachingRowLimit);

  ResultScanner scanner = TABLE.getScanner(scan);
  ClientScanner clientScanner = (ClientScanner) scanner;
  Result r = null;

  // Approximate the number of rows we expect will fit into the specified max rsult size. If this
  // approximation is less than caching, then we expect that the max result size limit will be
  // hit before the caching limit and thus partial results may be seen
  boolean expectToSeePartialResults = resultSizeRowLimit < cachingRowLimit;
  while ((r = clientScanner.next()) != null) {
    assertTrue(!r.isPartial() || expectToSeePartialResults);
  }

  scanner.close();
}

@Test( description = "Create valid batch instance and call batch method",
        dataProvider = "objCollection" )
public void testCreateValidBatchInstanceAndCallBatch( List<String> strCollection )
        throws Exception {

    final byte[] valQualifier = Bytes.toBytes("val");
    Batch.<String>newBuilder()
            .withBatchSize(3)
            .withObjectCollection(strCollection)
            .withMapper(string -> {
                final byte[] key = Bytes.toBytes(string);
                return new Put(key).addImmutable(TEST_CF_BYTES,
                                                 valQualifier,
                                                 key);
            })
            .withTable(testTable)
            .build()
            .call();

    final ResultScanner scanner = testTable.getScanner(TEST_CF_BYTES, valQualifier);
    final Integer rowCount
            = StreamSupport.stream(scanner.spliterator(), false)
                           .reduce(0,
                                   ( curVal, hresult ) -> curVal + hresult.size(),
                                   ( val1, val2 ) -> val1 + val2);
    assertThat(rowCount).isEqualTo(strCollection.size());
}

@Test( description = "Create valid batch instance and call batch method on Iterator",
        dataProvider = "objCollection" )
public void testCreateValidBatchInstanceAndCallBatchOnIterator( List<String> strCollection )
        throws Exception {

    final byte[] valQualifier = Bytes.toBytes("val");
    final Batch<String> batch = Batch.<String>newBuilder()
                                        .withBatchSize(3)
                                        .withObjectCollection(strCollection.iterator())
                                        .withMapper(string -> {
                                            final byte[] key = Bytes.toBytes(string);
                                            return new Put(key).addImmutable(TEST_CF_BYTES,
                                                                             valQualifier,
                                                                             key);
                                        })
                                        .withTable(testTable)
                                        .build();
    // batch on iterator instance perform only one time
    // other calls must have no effect(and must not fail)
    batch.call();
    batch.call();
    batch.call();

    final ResultScanner scanner = testTable.getScanner(TEST_CF_BYTES, valQualifier);
    final Integer rowCount
            = StreamSupport.stream(scanner.spliterator(), false)
                           .reduce(0,
                                   ( curVal, hresult ) -> curVal + hresult.size(),
                                   ( val1, val2 ) -> val1 + val2);
    assertThat(rowCount).isEqualTo(strCollection.size());
}

public List<T> extractData(ResultScanner results) throws Exception {
    List<T> rs = new ArrayList<>();
    int rowNum = 0;
    for (Result result : results) {
        rs.add(this.rowMapper.mapRow(result, rowNum++));
    }
    return rs;
}

protected final <T> T executeDistributedScan(TableName tableName, final Scan scan, final
AbstractRowKeyDistributor rowKeyDistributor, final ResultsExtractor<T> action) {
    assertAccessAvailable();
    return execute(tableName, new TableCallback<T>() {
        @Override
        public T doInTable(Table table) throws Throwable {
            StopWatch watch = null;
            if (debugEnabled) {
                watch = new StopWatch();
                watch.start();
            }
            final ResultScanner[] splitScanners = splitScan(table, scan, rowKeyDistributor);
            final ResultScanner scanner = new DistributedScanner(rowKeyDistributor,
                    splitScanners);
            if (debugEnabled) {
                logger.debug("DistributedScanner createTime: {}ms", watch.stop());
                watch.start();
            }
            try {
                return action.extractData(scanner);
            } finally {
                scanner.close();
                if (debugEnabled) {
                    logger.debug("DistributedScanner scanTime: {}ms", watch.stop());
                }
            }
        }
    });
}

private void closeScanner(ResultScanner[] scannerList) {
    for (ResultScanner scanner : scannerList) {
        if (scanner != null) {
            try {
                scanner.close();
            } catch (Exception e) {
                logger.warn("Scanner.close() error Caused:{}", e.getMessage(), e);
            }
        }
    }
}

public List<T> extractData(ResultScanner results) throws Exception {
    final List<T> rs = new ArrayList<>();
    int rowNum = 0;
    Result lastResult = null;

    for (Result result : results) {
        final T t = this.rowMapper.mapRow(result, rowNum);
        lastResult = result;
        if (t instanceof Collection) {
            rowNum += ((Collection<?>) t).size();
        } else if (t instanceof Map) {
            rowNum += ((Map<?, ?>) t).size();
        } else if (t == null) {
            // empty
        } else if (t.getClass().isArray()) {
            rowNum += Array.getLength(t);
        } else {
            rowNum++;
        }
        rs.add(t);
        if (rowNum >= limit) {
            break;
        }
    }

    eventHandler.handleLastResult(lastResult);
    return rs;
}

@Override
void testRow(int i) throws IOException {
  byte[] value = generateValue(this.rand);
  Scan scan = constructScan(value);
  ResultScanner scanner = null;
  try {
    scanner = this.table.getScanner(scan);
    while (scanner.next() != null) {
    }
  } finally {
    if (scanner != null) scanner.close();
  }
}

@Test
public void testPartialResultWhenRegionMove() throws IOException {
  Table table=createTestTable(TableName.valueOf("testPartialResultWhenRegionMove"),
      ROWS, FAMILIES, QUALIFIERS, VALUE);

  moveRegion(table, 1);

  Scan scan = new Scan();
  scan.setMaxResultSize(1);
  scan.setAllowPartialResults(true);
  ResultScanner scanner = table.getScanner(scan);
  for (int i = 0; i < NUM_FAMILIES * NUM_QUALIFIERS - 1; i++) {
    scanner.next();
  }
  Result result1 = scanner.next();
  assertEquals(1, result1.rawCells().length);
  Cell c1 = result1.rawCells()[0];
  assertCell(c1, ROWS[0], FAMILIES[NUM_FAMILIES - 1], QUALIFIERS[NUM_QUALIFIERS - 1]);
  assertFalse(result1.isPartial());

  moveRegion(table, 2);

  Result result2 = scanner.next();
  assertEquals(1, result2.rawCells().length);
  Cell c2 = result2.rawCells()[0];
  assertCell(c2, ROWS[1], FAMILIES[0], QUALIFIERS[0]);
  assertTrue(result2.isPartial());

  moveRegion(table, 3);

  Result result3 = scanner.next();
  assertEquals(1, result3.rawCells().length);
  Cell c3 = result3.rawCells()[0];
  assertCell(c3, ROWS[1], FAMILIES[0], QUALIFIERS[1]);
  assertTrue(result3.isPartial());

}