@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(); // Intersection consumes first cursor
    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(); // Intersection consumes second cursor as they come
    AtomicInteger falsePositives = new AtomicInteger();
    AsyncUtil.whileTrue(() -> intersectionCursor.onNext().thenApply(result -> {
        if (result.hasNext()) {
            int value = result.get();
            assertEquals(0, value % 3); // every result *must* be divisible by 3
            if (value % 2 != 0) {
                falsePositives.incrementAndGet(); // most results should be divisible by 2
            }
            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));
}