@Test public void longLists() { final Random r = new Random(0xba5eba11); for (int itr = 0; itr < 50; itr++) { long seed = r.nextLong(); LOGGER.info(KeyValueLogMessage.of("running intersection with large lists", TestLogMessageKeys.SEED, seed, TestLogMessageKeys.ITERATION, itr)); r.setSeed(seed); final List<List<Integer>> lists = Stream.generate( () -> IntStream.generate(() -> r.nextInt(500)).limit(1000).boxed().collect(Collectors.toList()) ).limit(5).collect(Collectors.toList()); final List<Function<byte[], RecordCursor<Integer>>> cursorFuncs = lists.stream() .map(list -> (Function<byte[], RecordCursor<Integer>>)((byte[] continuation) -> new RowLimitedCursor<>(RecordCursor.fromList(list, continuation), r.nextInt(50) + 10))) .collect(Collectors.toList()); final List<Set<Integer>> sets = lists.stream().map(HashSet::new).collect(Collectors.toList()); final Set<Integer> actualIntersection = new HashSet<>(sets.get(0)); sets.forEach(actualIntersection::retainAll); Set<Integer> found = new HashSet<>(); AtomicInteger falsePositives = new AtomicInteger(); boolean done = false; byte[] continuation = null; while (!done) { RecordCursor<Integer> intersectionCursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorFuncs, continuation, null); AsyncUtil.whileTrue(() -> intersectionCursor.onNext().thenApply(result -> { if (result.hasNext()) { int value = result.get(); assertThat(sets.stream().anyMatch(set -> set.contains(value)), is(true)); if (!actualIntersection.contains(value)) { falsePositives.incrementAndGet(); } found.add(value); } return result.hasNext(); }), intersectionCursor.getExecutor()).join(); RecordCursorResult<Integer> result = intersectionCursor.getNext(); assertThat(result.hasNext(), is(false)); if (result.getNoNextReason().isSourceExhausted()) { done = true; } else { assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, result.getNoNextReason()); } continuation = result.getContinuation().toBytes(); } assertThat(found.containsAll(actualIntersection), is(true)); LOGGER.info(KeyValueLogMessage.of("intersection false positives", "false_positives", falsePositives.get(), "actual_intersection_size", actualIntersection.size(), "iteration", itr)); assertThat(falsePositives.get(), lessThan(20)); } }
@Nonnull public static <T> ProbableIntersectionCursor<T> create( @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction, @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return create(comparisonKeyFunction, cursorFunctions, DEFAULT_EXPECTED_RESULTS, DEFAULT_FALSE_POSITIVE_PERCENTAGE, continuation, timer); }
ProbableIntersectionCursor extends MergeCursor<T, T, ProbableIntersectionCursorState<T>> { @Nonnull public static <T> ProbableIntersectionCursor<T> create( @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction, @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return create(comparisonKeyFunction, cursorFunctions, DEFAULT_EXPECTED_RESULTS, DEFAULT_FALSE_POSITIVE_PERCENTAGE, continuation, timer); } }
ProbableIntersectionCursor extends MergeCursor<T, T, ProbableIntersectionCursorState<T>> { @Nonnull public static <T> ProbableIntersectionCursor<T> create( @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction, @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return create(comparisonKeyFunction, cursorFunctions, DEFAULT_EXPECTED_RESULTS, DEFAULT_FALSE_POSITIVE_PERCENTAGE, continuation, timer); } ProbableIntersectionCursor(@Nonnull List<ProbableIntersectionCursorState<T>> cursorStates, @Nullable FDBStoreTimer timer); }
ProbableIntersectionCursor extends MergeCursor<T, T, ProbableIntersectionCursorState<T>> { @Nonnull public static <T> ProbableIntersectionCursor<T> create( @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction, @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return create(comparisonKeyFunction, cursorFunctions, DEFAULT_EXPECTED_RESULTS, DEFAULT_FALSE_POSITIVE_PERCENTAGE, continuation, timer); } ProbableIntersectionCursor(@Nonnull List<ProbableIntersectionCursorState<T>> cursorStates, @Nullable FDBStoreTimer timer); @Nonnull static ProbableIntersectionCursor<T> create(
            @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction,
            @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions,
            @Nullable byte[] continuation,
            @Nullable FDBStoreTimer timer); @Nonnull static ProbableIntersectionCursor<T> create(
            @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction,
            @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions,
            long expectedResults,
            double falsePositivePercentage,
            @Nullable byte[] continuation,
            @Nullable FDBStoreTimer timer); }
ProbableIntersectionCursor extends MergeCursor<T, T, ProbableIntersectionCursorState<T>> { @Nonnull public static <T> ProbableIntersectionCursor<T> create( @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction, @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return create(comparisonKeyFunction, cursorFunctions, DEFAULT_EXPECTED_RESULTS, DEFAULT_FALSE_POSITIVE_PERCENTAGE, continuation, timer); } ProbableIntersectionCursor(@Nonnull List<ProbableIntersectionCursorState<T>> cursorStates, @Nullable FDBStoreTimer timer); @Nonnull static ProbableIntersectionCursor<T> create(
            @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction,
            @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions,
            @Nullable byte[] continuation,
            @Nullable FDBStoreTimer timer); @Nonnull static ProbableIntersectionCursor<T> create(
            @Nonnull Function<? super T, ? extends List<Object>> comparisonKeyFunction,
            @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions,
            long expectedResults,
            double falsePositivePercentage,
            @Nullable byte[] continuation,
            @Nullable FDBStoreTimer timer); static final long DEFAULT_EXPECTED_RESULTS; static final double DEFAULT_FALSE_POSITIVE_PERCENTAGE; }