@Test public void resumeFromContinuation() { final FDBStoreTimer timer = new FDBStoreTimer(); final List<Integer> list1 = Arrays.asList(10, 2, 5, 6, 8, 19, 0); final List<Integer> list2 = Arrays.asList( 9, 1, 3, 5, 2, 4, 8); final List<Function<byte[], RecordCursor<Integer>>> cursorFuncs = listsToFunctions(Arrays.asList(list1, list2)); final Function<byte[], ProbableIntersectionCursor<Integer>> intersectionCursorFunction = continuation -> ProbableIntersectionCursor.create(Collections::singletonList, cursorFuncs, continuation, timer); final Iterator<Integer> resultIterator = Iterators.forArray(5, 2, 8); byte[] continuation = null; boolean done = false; List<BloomFilter<List<Object>>> lastBloomFilters = null; while (!done) { ProbableIntersectionCursor<Integer> intersectionCursor = intersectionCursorFunction.apply(continuation); List<BloomFilter<List<Object>>> bloomFilters = intersectionCursor.getCursorStates().stream() .map(ProbableIntersectionCursorState::getBloomFilter) .collect(Collectors.toList()); if (lastBloomFilters != null) { assertEquals(lastBloomFilters, bloomFilters); } lastBloomFilters = bloomFilters; RecordCursorResult<Integer> result = intersectionCursor.getNext(); if (resultIterator.hasNext()) { assertThat(result.hasNext(), is(true)); assertEquals(resultIterator.next(), result.get()); assertThat(result.getContinuation().isEnd(), is(false)); assertNotNull(result.getContinuation().toBytes()); } else { assertThat(result.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason()); assertThat(result.getContinuation().isEnd(), is(true)); assertNull(result.getContinuation().toBytes()); done = true; } continuation = result.getContinuation().toBytes(); } assertEquals(3, timer.getCount(FDBStoreTimer.Counts.QUERY_INTERSECTION_PLAN_MATCHES)); assertEquals(list1.size() + list2.size() - 3, timer.getCount(FDBStoreTimer.Counts.QUERY_INTERSECTION_PLAN_NONMATCHES)); }
@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; }