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; }
@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)); }
@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)); }
@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)); } }
@Test public void noNextReasons() { RecordCursor<Integer> cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( new RecordCursorTest.FakeOutOfBandCursor<>(RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)), 3), new RecordCursorTest.FakeOutOfBandCursor<>(RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)), 2) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.TIME_LIMIT_REACHED, 3); cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( new RecordCursorTest.FakeOutOfBandCursor<>(RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)), 3), RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)).limitRowsTo(2) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.TIME_LIMIT_REACHED, 3); cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)).limitRowsTo(3), RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)).limitRowsTo(2) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, 3); cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( new RecordCursorTest.FakeOutOfBandCursor<>(RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)), 3), RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.TIME_LIMIT_REACHED, 3, 4, 1); cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)).limitRowsTo(3), RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, 3, 4, 1); cursor = ProbableIntersectionCursor.create(Collections::singletonList, cursorsToFunctions(Arrays.asList( RecordCursor.fromList(Arrays.asList(1, 4, 3, 7, 9)), RecordCursor.fromList(Arrays.asList(3, 7, 8, 4, 1)) )), null, null); verifyResults(cursor, RecordCursor.NoNextReason.SOURCE_EXHAUSTED, 3, 7, 4, 1); }
@Test public void errorInChild() { CompletableFuture<Integer> future = new CompletableFuture<>(); RecordCursor<Integer> cursor = ProbableIntersectionCursor.create(Collections::singletonList, Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation), continuation -> RecordCursor.fromFuture(future) ), null, null); CompletableFuture<RecordCursorResult<Integer>> cursorResultFuture = cursor.onNext(); final RecordCoreException ex = new RecordCoreException("something bad happened!"); future.completeExceptionally(ex); ExecutionException executionException = assertThrows(ExecutionException.class, cursorResultFuture::get); assertNotNull(executionException.getCause()); assertSame(ex, executionException.getCause()); }
@Test public void errorAndLimitInChild() { CompletableFuture<Integer> future = new CompletableFuture<>(); RecordCursor<Integer> cursor = ProbableIntersectionCursor.create(Collections::singletonList, Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation).limitRowsTo(1), continuation -> RecordCursor.fromFuture(future) ), null, null); CompletableFuture<RecordCursorResult<Integer>> cursorResultFuture = cursor.onNext(); final RecordCoreException ex = new RecordCoreException("something bad happened!"); future.completeExceptionally(ex); ExecutionException executionException = assertThrows(ExecutionException.class, cursorResultFuture::get); assertNotNull(executionException.getCause()); assertSame(ex, executionException.getCause()); }
@Test public void loopIterationWithLimit() throws ExecutionException, InterruptedException { FDBStoreTimer timer = new FDBStoreTimer(); FirableCursor<Integer> secondCursor = new FirableCursor<>(RecordCursor.fromList(Arrays.asList(2, 1))); RecordCursor<Integer> cursor = ProbableIntersectionCursor.create(Collections::singletonList, Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation).limitRowsTo(1), continuation -> secondCursor ), null, timer); CompletableFuture<RecordCursorResult<Integer>> cursorResultFuture = cursor.onNext(); secondCursor.fire(); assertFalse(cursorResultFuture.isDone()); secondCursor.fire(); RecordCursorResult<Integer> cursorResult = cursorResultFuture.get(); assertEquals(1, (int)cursorResult.get()); secondCursor.fire(); cursorResult = cursor.getNext(); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursorResult.getNoNextReason()); assertThat(timer.getCount(FDBStoreTimer.Events.QUERY_INTERSECTION), lessThanOrEqualTo(5)); }