HBaseOutput implements RandomOutput, BulkOutput, ProvidesAlias, ProvidesValidations { @Override public Iterable<Row> getExistingForFilters(Iterable<Row> filters) throws Exception { LOG.debug("Fetching filter rows from table: {}", tableName.toString()); List<Row> filterResults = Lists.newArrayList(); try (Table table = getConnection(config).getTable(tableName)) { List<Get> gets = Lists.newArrayList(); List<Scan> scans = Lists.newArrayList(); for (Row filter : filters) { Query query = getSerde(config).convertToQuery(filter); LOG.debug("Adding filter: {}", query); if (query instanceof Get) { gets.add((Get)query); } else if (query instanceof Scan) { scans.add((Scan)query); } else { throw new RuntimeException("Unsupported HBase query class: " + query.getClass().getName()); } } List<Result> results = Lists.newArrayList(); if (gets.size() > 0) { results.addAll(Lists.newArrayList(table.get(gets))); } if (scans.size() > 0) { Scan mergedScan = HBaseUtils.mergeRangeScans(scans); results.addAll(Lists.newArrayList(table.getScanner(mergedScan))); } filterResults.addAll(getSerde(config).convertFromResults(results)); } return filterResults; }  @Override void configure(Config config); @Override Iterable<Row> getExistingForFilters(Iterable<Row> filters); @Override void applyRandomMutations(List<Row> plannedRows); @Override Set<MutationType> getSupportedRandomMutationTypes(); @Override Set<MutationType> getSupportedBulkMutationTypes(); @Override void applyBulkMutations(List<Tuple2<MutationType, Dataset<Row>>> planned); @Override String getAlias(); @Override Validations getValidations();  }
@Test public void testGetExistingForFilters() throws Exception { addEntriesToHBase(); Table table = connection.getTable(TableName.valueOf(TABLE)); scanAndCountTable(table, INPUT_ROWS * 4); Config config = ConfigUtils.configFromResource("/hbase/hbase-output-test.conf").getConfig("output"); config = config.withValue("zookeeper", ConfigValueFactory.fromAnyRef("localhost:" + utility.getZkCluster().getClientPort())); HBaseOutput output = new HBaseOutput(); output.configure(config); List<Row> filters = Lists.newArrayList(); filters.add(new RowWithSchema(filterSchema, "AAPL", 1_000_000_000L)); filters.add(new RowWithSchema(filterSchema, "GOOG", 1_000_000_004L)); Iterable<Row> filtered = output.getExistingForFilters(filters); assertEquals(2, Iterables.size(filtered)); Row row1 = Iterables.get(filtered, 0); Row row2 = Iterables.get(filtered, 1); if (!row1.get(row1.fieldIndex("symbol")).equals("GOOG")) { Row tmp = row1; row1 = row2; row2 = tmp; } assertEquals(6, row1.schema().length()); assertEquals("GOOG", row1.get(row1.fieldIndex("symbol"))); assertEquals(1_000_000_004L, (long) row1.get(row1.fieldIndex("transacttime"))); assertEquals("4", row1.get(row1.fieldIndex("clordid"))); assertEquals(4, (int) row1.get(row1.fieldIndex("orderqty"))); assertEquals(4, (int) row1.get(row1.fieldIndex("leavesqty"))); assertEquals(8, (int) row1.get(row1.fieldIndex("cumqty"))); assertEquals(6, row2.schema().length()); assertEquals("AAPL", row2.get(row2.fieldIndex("symbol"))); assertEquals(1_000_000_000L, row2.get(row2.fieldIndex("transacttime"))); assertEquals("0", row2.get(row2.fieldIndex("clordid"))); assertEquals(0, (int) row2.get(row2.fieldIndex("orderqty"))); assertEquals(0, (int) row2.get(row2.fieldIndex("leavesqty"))); assertEquals(0, (int) row2.get(row2.fieldIndex("cumqty"))); }