@Test public void basicIntersection() { final FDBStoreTimer timer = new FDBStoreTimer(); final Iterator<Integer> iterator1 = IntStream.iterate(0, x -> x + 2).limit(150).iterator(); final Iterator<Integer> iterator2 = IntStream.iterate(0, x -> x + 3).limit(100).iterator(); final FirableCursor<Integer> cursor1 = new FirableCursor<>(RecordCursor.fromIterator(iterator1)); final FirableCursor<Integer> cursor2 = new FirableCursor<>(RecordCursor.fromIterator(iterator2)); final RecordCursor<Integer> intersectionCursor = ProbableIntersectionCursor.create( Collections::singletonList, Arrays.asList(bignore -> cursor1, bignore -> cursor2), null, timer ); cursor1.fireAll(); CompletableFuture<RecordCursorResult<Integer>> firstFuture = intersectionCursor.onNext(); cursor2.fire(); RecordCursorResult<Integer> firstResult = firstFuture.join(); assertEquals(0, (int)firstResult.get()); assertThat(firstResult.hasNext(), is(true)); assertEquals(cursor1.getNext().getNoNextReason(), RecordCursor.NoNextReason.SOURCE_EXHAUSTED); cursor2.fireAll(); AtomicInteger falsePositives = new AtomicInteger(); AsyncUtil.whileTrue(() -> intersectionCursor.onNext().thenApply(result -> { if (result.hasNext()) { int value = result.get(); assertEquals(0, value % 3); if (value % 2 != 0) { falsePositives.incrementAndGet(); } assertThat(result.getContinuation().isEnd(), is(false)); assertNotNull(result.getContinuation().toBytes()); try { RecordCursorProto.ProbableIntersectionContinuation protoContinuation = RecordCursorProto.ProbableIntersectionContinuation.parseFrom(result.getContinuation().toBytes()); assertEquals(2, protoContinuation.getChildStateCount()); assertThat(protoContinuation.getChildState(0).getExhausted(), is(true)); assertThat(protoContinuation.getChildState(0).hasContinuation(), is(false)); assertThat(protoContinuation.getChildState(1).getExhausted(), is(false)); assertThat(protoContinuation.getChildState(1).hasContinuation(), is(true)); } catch (InvalidProtocolBufferException e) { throw new RecordCoreException("error parsing proto continuation", e); } } else { assertThat(result.getNoNextReason().isSourceExhausted(), is(true)); assertThat(result.getContinuation().isEnd(), is(true)); assertNull(result.getContinuation().toBytes()); } return result.hasNext(); }), intersectionCursor.getExecutor()).join(); assertThat(falsePositives.get(), lessThan(5)); assertEquals(50 + falsePositives.get(), timer.getCount(FDBStoreTimer.Counts.QUERY_INTERSECTION_PLAN_MATCHES)); assertEquals(200 - falsePositives.get(), 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; }