AbstractSampleElementsForSplitPointsHandler implements OutputOperationHandler<SampleElementsForSplitPoints<T>, List<T>> { @Override public List<T> doOperation(final SampleElementsForSplitPoints<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 float proportionToSample = operation.getProportionToSample(); final Random random = new Random(System.currentTimeMillis()); final Iterable<? extends Element> cleanElements = Iterables.filter( operation.getInput(), e -> null != e && (1 == proportionToSample || random.nextFloat() <= proportionToSample) ); final LimitedCloseableIterable<? extends Element> limitedElements = new LimitedCloseableIterable<>(cleanElements, 0, maxSampledElements, false); final Stream<T> recordStream = process(Streams.toStream(limitedElements), typedStore); final Stream<T> sortedRecordStream = sort(recordStream, typedStore); final List<T> records = sortedRecordStream.collect(Collectors.toList()); return store.execute( new GenerateSplitPointsFromSample.Builder<T>() .input(records) .numSplits(numSplits) .build(), context); }  @Override List<T> doOperation(final SampleElementsForSplitPoints<T> operation, final Context context, final Store store); int getMaxSampledElements(); void setMaxSampledElements(final int maxSampledElements); static final int MAX_SAMPLED_ELEMENTS_DEFAULT; }
@Test public void shouldThrowExceptionForNullInput() throws OperationException { final AbstractSampleElementsForSplitPointsHandler<?, S> handler = createHandler(); final SampleElementsForSplitPoints operation = new SampleElementsForSplitPoints.Builder<>() .numSplits(1) .build(); try { handler.doOperation(operation, new Context(), createStore()); fail("Exception expected"); } catch (final OperationException e) { assertTrue(e.getMessage().contains("input is required"), e.getMessage()); } }
@Test public void shouldUseFullSampleOfAllElementsByDefault() throws OperationException { final int numSplits = 3; final List<Element> elements = IntStream.range(0, numSplits) .mapToObj(i -> new Entity(TestGroups.ENTITY, "vertex_" + i)) .collect(Collectors.toList()); final AbstractSampleElementsForSplitPointsHandler<?, S> handler = createHandler(); final SampleElementsForSplitPoints operation = new SampleElementsForSplitPoints.Builder<>() .input(elements) .numSplits(numSplits) .build(); final S store = createStore(); handler.doOperation(operation, new Context(), store); final ArgumentCaptor<GenerateSplitPointsFromSample> generateSplitPointsFromSampleCaptor = ArgumentCaptor.forClass(GenerateSplitPointsFromSample.class); verify(store).execute(generateSplitPointsFromSampleCaptor.capture(), any(Context.class)); assertExpectedNumberOfSplitPointsAndSampleSize(generateSplitPointsFromSampleCaptor, numSplits, elements.size()); }
@Test public void shouldFilterOutNulls() throws OperationException { final int numSplits = 3; final List<Element> elements = IntStream.range(0, numSplits) .mapToObj(i -> new Entity(TestGroups.ENTITY, "vertex_" + i)) .collect(Collectors.toList()); final List<Element> elementsWithNulls = new ArrayList<>(); elementsWithNulls.add(null); elementsWithNulls.addAll(elements); elementsWithNulls.add(null); final AbstractSampleElementsForSplitPointsHandler<?, S> handler = createHandler(); final SampleElementsForSplitPoints operation = new SampleElementsForSplitPoints.Builder<>() .input(elementsWithNulls) .numSplits(numSplits) .build(); final S store = createStore(); handler.doOperation(operation, new Context(), store); final ArgumentCaptor<GenerateSplitPointsFromSample> generateSplitPointsFromSampleCaptor = ArgumentCaptor.forClass(GenerateSplitPointsFromSample.class); verify(store).execute(generateSplitPointsFromSampleCaptor.capture(), any(Context.class)); assertExpectedNumberOfSplitPointsAndSampleSize(generateSplitPointsFromSampleCaptor, numSplits, elements.size()); }
@Test public void shouldSampleApproximatelyHalfOfElements() throws OperationException { final int numSplits = 3; final List<Element> elements = IntStream.range(0, 1000 * numSplits) .mapToObj(i -> new Entity(TestGroups.ENTITY, "vertex_" + i)) .collect(Collectors.toList()); final AbstractSampleElementsForSplitPointsHandler<?, S> handler = createHandler(); final SampleElementsForSplitPoints operation = new SampleElementsForSplitPoints.Builder<>() .input(elements) .numSplits(numSplits) .proportionToSample(0.5f) .build(); final S store = createStore(); handler.doOperation(operation, new Context(), store); final ArgumentCaptor<GenerateSplitPointsFromSample> generateSplitPointsFromSampleCaptor = ArgumentCaptor.forClass(GenerateSplitPointsFromSample.class); verify(store).execute(generateSplitPointsFromSampleCaptor.capture(), any(Context.class)); final int maximumExpectedSampleSize = (int) ((elements.size() / 2) * 1.1); assertExpectedNumberOfSplitPointsAndSampleSizeOfNoMoreThan(generateSplitPointsFromSampleCaptor, numSplits, maximumExpectedSampleSize); }