AbstractGenerateSplitPointsFromSampleHandler implements OutputOperationHandler<GenerateSplitPointsFromSample<T>, List<T>> { @Override public List<T> doOperation(final GenerateSplitPointsFromSample<T> operation, final Context context, final Store store) throws OperationException { final S typedStore = (S) store; validate(operation, typedStore); final Integer numSplits = getNumSplits(operation, typedStore); if (null == numSplits) { throw new OperationException("Number of splits is required"); } if (numSplits < 1) { return Collections.emptyList(); } final List<T> records = Streams.toStream(operation.getInput()).collect(Collectors.toList()); final List<T> splits; if (records.size() < 2 || records.size() <= numSplits) { splits = records; } else { final LinkedHashSet<T> splitsSet = Integer.MAX_VALUE != numSplits ? new LinkedHashSet<>(numSplits) : new LinkedHashSet<>(); final double outputEveryNthRecord = ((double) records.size()) / (numSplits + 1); int nthCount = 0; for (final T record : records) { nthCount++; if (nthCount >= (int) (outputEveryNthRecord * (splitsSet.size() + 1))) { splitsSet.add(record); if (numSplits == splitsSet.size()) { break; } } } splits = new ArrayList<>(splitsSet); } return splits; }  @Override List<T> doOperation(final GenerateSplitPointsFromSample<T> operation, final Context context, final Store store);  }
@Test public void shouldThrowExceptionForNullInput() throws OperationException { final AbstractGenerateSplitPointsFromSampleHandler<?, S> handler = createHandler(); final GenerateSplitPointsFromSample operation = new GenerateSplitPointsFromSample.Builder<>() .numSplits(1) .build(); OperationException actual = assertThrows(OperationException.class, () -> handler.doOperation(operation, new Context(), createStore())); assertTrue(actual.getMessage().contains("input is required")); }
@Test public void shouldReturnEmptyCollectionIfNumSplitsIsLessThan1() throws OperationException { final List<String> sample = createSampleOfSize(100); final AbstractGenerateSplitPointsFromSampleHandler<?, S> handler = createHandler(); final GenerateSplitPointsFromSample operation = new GenerateSplitPointsFromSample.Builder<>() .input(sample) .numSplits(0) .build(); final List<?> splits = handler.doOperation(operation, new Context(), createStore()); assertTrue(splits.isEmpty()); }
@Test public void shouldCalculateRequiredNumberOfSplits() throws OperationException { final int numSplits = 3; final List<String> sample = createSampleOfSize(numSplits * 10); final AbstractGenerateSplitPointsFromSampleHandler<?, S> handler = createHandler(); final GenerateSplitPointsFromSample operation = new GenerateSplitPointsFromSample.Builder<>() .input(sample) .numSplits(numSplits) .build(); final List<?> splits = handler.doOperation(operation, new Context(), createStore()); verifySplits(Arrays.asList(6, 14, 21), sample, splits, handler); }
@Test public void shouldDeduplicateElements() throws OperationException { final int numSplits = 3; final List<String> sample = Collections.nCopies(numSplits * 10, "key1"); final AbstractGenerateSplitPointsFromSampleHandler<?, S> handler = createHandler(); final GenerateSplitPointsFromSample operation = new GenerateSplitPointsFromSample.Builder<>() .input(sample) .numSplits(numSplits) .build(); final S store = createStore(); final List<?> splits = handler.doOperation(operation, new Context(), createStore()); assertEquals(1, splits.size()); verifySplits(Collections.singletonList(0), sample, splits, handler); }