@Test public void testDenseSlicedTiles () throws Exception { TileIndex index = new TileIndex(0, 0, 0, 4, 4); TileData<List<Integer>> tile = new DenseTileData<>(index); for (int x=0; x<4; ++x) { for (int y=0; y<4; ++y) { List<Integer> bin = new ArrayList<>(); for (int z=0; z<4; ++z) { bin.add(x+y+z); } tile.setBin(x, y, bin); } } String table = "dense-bucket-test"; HBaseSlicedPyramidIO io = new HBaseSlicedPyramidIO("hadoop-s1", "2181", "hadoop-s1:60000"); TileSerializer<List<Integer>> serializer = new KryoSerializer<>(new TypeDescriptor(List.class, new TypeDescriptor(Integer.class))); io.initializeForWrite(table); io.writeTiles(table, serializer, Arrays.asList(tile)); List<TileIndex> indices = Arrays.asList(index); TileData<List<Integer>> full = io.readTiles(table, serializer, indices).get(0); TileData<List<Integer>> s0 = io.readTiles(table+"[0]", serializer, indices).get(0); TileData<List<Integer>> s1 = io.readTiles(table+"[1]", serializer, indices).get(0); TileData<List<Integer>> s2 = io.readTiles(table+"[2]", serializer, indices).get(0); TileData<List<Integer>> s3 = io.readTiles(table+"[3]", serializer, indices).get(0); List<TileData<List<Integer>>> slices = Arrays.asList(s0, s1, s2, s3); TileData<List<Integer>> s01 = io.readTiles(table+"[0-1]", serializer, indices).get(0); TileData<List<Integer>> s23 = io.readTiles(table+"[2-3]", serializer, indices).get(0); List<TileData<List<Integer>>> slicePairs = Arrays.asList(s01, s23); for (int x=0; x<4; ++x) { for (int y=0; y<4; ++y) { for (int z=0; z<4; ++z) { Assert.assertEquals(tile.getBin(x, y).get(z), full.getBin(x, y).get(z)); Assert.assertEquals(tile.getBin(x, y).get(z), slices.get(z).getBin(x, y).get(0)); Assert.assertEquals(tile.getBin(x, y).get(z), slicePairs.get(z/2).getBin(x, y).get(z%2)); } } } }
@Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); @Override HBaseTilePutter getPutter(); void setPyramidding(boolean doPyramidding); static HBaseColumn getSliceColumn(int minSlice, int maxSlice); @Override List<TileData<T>> readTiles(String tableName,
											TileSerializer<T> serializer,
											Iterable<TileIndex> tiles); static List<Pair<Integer, Integer>> decomposeRange(int start, int end); }
HBaseSlicedPyramidIO extends HBasePyramidIO { @Override public <T> List<TileData<T>> readTiles (String tableName, TileSerializer<T> serializer, Iterable<TileIndex> tiles) throws IOException { Matcher m = SLICE_PATTERN.matcher(tableName); TypeDescriptor binType = serializer.getBinTypeDescription(); if (List.class == binType.getMainType() && m.matches()) { String realName = m.group("table"); HBaseColumn[] columns; int min = Integer.parseInt(m.group("min")); if (null == m.group("max")) { columns = new HBaseColumn[] { getSliceColumn(min, min) }; } else { int max = Integer.parseInt(m.group("max")); List<Pair<Integer, Integer>> sliceRanges; if (_doPyramidding) { sliceRanges = decomposeRange(min, max); } else { sliceRanges = new ArrayList<>(); for (int n=min; n<=max; ++n) { sliceRanges.add(new Pair<>(n, n)); } } columns = new HBaseColumn[sliceRanges.size()]; for (int i = 0; i < sliceRanges.size(); ++i) { Pair<Integer, Integer> sliceRange = sliceRanges.get(i); columns[i] = getSliceColumn(sliceRange.getFirst(), sliceRange.getSecond()); } } List<TileData<T>> rawResults = super.readTiles(realName, serializer, tiles, columns); if (1 == columns.length) return rawResults; else { int numRaw = rawResults.size(); int numReal = numRaw/columns.length; List<TileData<T>> realResults = new ArrayList<>(numReal); for (int i=0; i<numReal; ++i) { realResults.add(compose((List) rawResults, i, columns.length, numReal)); } return realResults; } } else { return super.readTiles(tableName, serializer, tiles); } } HBaseSlicedPyramidIO(String zookeeperQuorum, String zookeeperPort, String hbaseMaster); @Override HBaseTilePutter getPutter(); void setPyramidding(boolean doPyramidding); static HBaseColumn getSliceColumn(int minSlice, int maxSlice); @Override List<TileData<T>> readTiles(String tableName,
											TileSerializer<T> serializer,
											Iterable<TileIndex> tiles); static List<Pair<Integer, Integer>> decomposeRange(int start, int end);  }