BundleSplitter implements Serializable { List<OffsetRange> getBundleSizes(int desiredNumBundles, long start, long end) { List<OffsetRange> result = new ArrayList<>(); double[] relativeSizes = getRelativeBundleSizes(desiredNumBundles); double s = sum(relativeSizes); long startOffset = start; double sizeSoFar = 0; for (int i = 0; i < relativeSizes.length; ++i) { sizeSoFar += relativeSizes[i]; long endOffset = (i == relativeSizes.length - 1) ? end : (long) (start + sizeSoFar * (end - start) / s); if (startOffset != endOffset) { result.add(new OffsetRange(startOffset, endOffset)); } startOffset = endOffset; } return result; } BundleSplitter(SyntheticSourceOptions options);   }
@Test public void shouldCreateAsManyBundlesAsItsDesired() { splitter = new BundleSplitter(options); int expectedBundleCount = 4; List bundleSizes = splitter.getBundleSizes(expectedBundleCount, 0, options.numRecords); assertEquals(expectedBundleCount, bundleSizes.size()); }
@Test public void bundlesShouldBeEvenForConstDistribution() { long expectedBundleSize = 2; options.bundleSizeDistribution = fromRealDistribution(new ConstantRealDistribution(2)); splitter = new BundleSplitter(options); List<OffsetRange> bundleSizes = splitter.getBundleSizes(4, 0, options.numRecords); bundleSizes.stream() .map(range -> range.getTo() - range.getFrom()) .forEach(size -> assertEquals(expectedBundleSize, size.intValue())); }
@Test public void bundleSizesShouldBeProportionalToTheOneSuggestedInBundleSizeDistribution() { long expectedBundleSize = 4; options.bundleSizeDistribution = fromRealDistribution(new ConstantRealDistribution(2)); options.numRecords = 16; splitter = new BundleSplitter(options); List<OffsetRange> bundleSizes = splitter.getBundleSizes(4, 0, options.numRecords); bundleSizes.stream() .map(range -> range.getTo() - range.getFrom()) .forEach(size -> assertEquals(expectedBundleSize, size.intValue())); }
@Test public void consequentBundlesShouldHaveTheSameRangeEndAndStart() { int desiredNumberOfBundles = 2; options.bundleSizeDistribution = fromRealDistribution(new ConstantRealDistribution(2)); splitter = new BundleSplitter(options); List<OffsetRange> bundleSizes = splitter.getBundleSizes(desiredNumberOfBundles, 0, options.numRecords); assertEquals(bundleSizes.get(0).getTo(), bundleSizes.get(1).getFrom()); assertEquals(bundleSizes.get(0).getTo(), bundleSizes.get(1).getFrom()); assertEquals(desiredNumberOfBundles, bundleSizes.size()); }