HBaseIO { public static Read read() { return new Read(null, "", new Scan()); } private  HBaseIO(); static Read read(); @Experimental(Kind.SPLITTABLE_DO_FN) static ReadAll readAll(); static Write write();  }
@Test public void testReadBuildsCorrectly() { HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId("table"); assertEquals("table", read.getTableId()); assertNotNull("configuration", read.getConfiguration()); }
@Test public void testReadBuildsCorrectlyInDifferentOrder() { HBaseIO.Read read = HBaseIO.read().withTableId("table").withConfiguration(conf); assertEquals("table", read.getTableId()); assertNotNull("configuration", read.getConfiguration()); }
@Test public void testReadingFailsTableDoesNotExist() { final String table = tmpTable.getName(); thrown.expect(IllegalArgumentException.class); thrown.expectMessage(String.format("Table %s does not exist", table)); runReadTest( HBaseIO.read().withConfiguration(conf).withTableId(table), false, new ArrayList<>()); runReadTest(HBaseIO.read().withConfiguration(conf).withTableId(table), true, new ArrayList<>()); }
@Test public void testReadingEmptyTable() throws Exception { final String table = tmpTable.getName(); createTable(table); runReadTest( HBaseIO.read().withConfiguration(conf).withTableId(table), false, new ArrayList<>()); runReadTest(HBaseIO.read().withConfiguration(conf).withTableId(table), true, new ArrayList<>()); }
@Test public void testReading() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; createAndWriteData(table, numRows); runReadTestLength(HBaseIO.read().withConfiguration(conf).withTableId(table), false, numRows); }
@Test public void testReadingSDF() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; createAndWriteData(table, numRows); runReadTestLength(HBaseIO.read().withConfiguration(conf).withTableId(table), true, numRows); }
@Test public void testReadingWithSplits() throws Exception { final String table = tmpTable.getName(); final int numRows = 1500; final int numRegions = 4; final long bytesPerRow = 100L; createAndWriteData(table, numRows); HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId(table); HBaseSource source = new HBaseSource(read, null ); List<? extends BoundedSource<Result>> splits = source.split(numRows * bytesPerRow / numRegions, null ); assertThat(splits, hasSize(4)); assertSourcesEqualReferenceSource(source, splits, null ); }
@Test public void testReadingSourceTwice() throws Exception { final String table = tmpTable.getName(); final int numRows = 10; createAndWriteData(table, numRows); HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId(table); HBaseSource source = new HBaseSource(read, null ); assertThat(SourceTestUtils.readFromSource(source, null), hasSize(numRows)); assertThat(SourceTestUtils.readFromSource(source, null), hasSize(numRows)); }
@Test public void testReadingWithFilter() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; createAndWriteData(table, numRows); String regex = ".*17.*"; Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regex)); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withFilter(filter), false, 20); }
@Test public void testReadingWithFilterSDF() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; createAndWriteData(table, numRows); String regex = ".*17.*"; Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regex)); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withFilter(filter), true, 20); }
@Test public void testReadingKeyRangePrefix() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; final ByteKey startKey = ByteKey.copyFrom("2".getBytes(StandardCharsets.UTF_8)); createAndWriteData(table, numRows); final ByteKeyRange prefixRange = ByteKeyRange.ALL_KEYS.withEndKey(startKey); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withKeyRange(prefixRange), false, 126); }
@Test public void testReadingKeyRangeSuffix() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; final ByteKey startKey = ByteKey.copyFrom("2".getBytes(StandardCharsets.UTF_8)); createAndWriteData(table, numRows); final ByteKeyRange suffixRange = ByteKeyRange.ALL_KEYS.withStartKey(startKey); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withKeyRange(suffixRange), false, 875); }
@Test public void testReadingKeyRangeMiddle() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; final byte[] startRow = "2".getBytes(StandardCharsets.UTF_8); final byte[] stopRow = "9".getBytes(StandardCharsets.UTF_8); createAndWriteData(table, numRows); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withKeyRange(startRow, stopRow), false, 441); }
@Test public void testReadingKeyRangeMiddleSDF() throws Exception { final String table = tmpTable.getName(); final int numRows = 1001; final byte[] startRow = "2".getBytes(StandardCharsets.UTF_8); final byte[] stopRow = "9".getBytes(StandardCharsets.UTF_8); createAndWriteData(table, numRows); runReadTestLength( HBaseIO.read().withConfiguration(conf).withTableId(table).withKeyRange(startRow, stopRow), true, 441); }
@Test public void testReadingSplitAtFractionExhaustive() throws Exception { final String table = tmpTable.getName(); final int numRows = 7; createAndWriteData(table, numRows); HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId(table); HBaseSource source = new HBaseSource(read, null ) .withStartKey(ByteKey.of(48)) .withEndKey(ByteKey.of(58)); assertSplitAtFractionExhaustive(source, null); }
@Test public void testReadingSplitAtFraction() throws Exception { final String table = tmpTable.getName(); final int numRows = 10; createAndWriteData(table, numRows); HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId(table); HBaseSource source = new HBaseSource(read, null ); double k = 0.003922; assertSplitAtFractionFails(source, 0, k, null ); assertSplitAtFractionFails(source, 0, 1.0, null ); assertSplitAtFractionSucceedsAndConsistent(source, 1, k, null ); assertSplitAtFractionSucceedsAndConsistent(source, 1, 0.666, null ); assertSplitAtFractionFails(source, 3, 2 * k, null ); assertSplitAtFractionSucceedsAndConsistent(source, 3, 3 * k, null ); assertSplitAtFractionSucceedsAndConsistent(source, 3, 4 * k, null ); assertSplitAtFractionFails(source, 6, 5 * k, null ); assertSplitAtFractionSucceedsAndConsistent(source, 6, 0.7, null ); }
@Test public void testReadingDisplayData() { HBaseIO.Read read = HBaseIO.read().withConfiguration(conf).withTableId("fooTable"); DisplayData displayData = DisplayData.from(read); assertThat(displayData, hasDisplayItem("tableId", "fooTable")); assertThat(displayData, hasDisplayItem("configuration")); }