UnorderedUnionCursor extends UnionCursorBase<T, MergeCursorState<T>> { @Nonnull public static <T> UnorderedUnionCursor<T> create( @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions, @Nullable byte[] continuation, @Nullable FDBStoreTimer timer) { return new UnorderedUnionCursor<>(createCursorStates(cursorFunctions, continuation), timer); } protected  UnorderedUnionCursor(@Nonnull List<MergeCursorState<T>> cursorStates,
                                   @Nullable FDBStoreTimer timer); @Nonnull static UnorderedUnionCursor<T> create(
            @Nonnull List<Function<byte[], RecordCursor<T>>> cursorFunctions,
            @Nullable byte[] continuation,
            @Nullable FDBStoreTimer timer);  }
@Test public void basicUnion() throws ExecutionException, InterruptedException { final List<List<Integer>> elems = Arrays.asList( Arrays.asList(0, 100, 200), Arrays.asList(401, 201, 1), Arrays.asList(2, 302, 102) ); final UnorderedUnionCursor<Integer> cursor = UnorderedUnionCursor.create(functionsFromLists(elems), null, null); List<Integer> results = cursor.asList().get(); assertEquals(elems.stream().mapToInt(List::size).sum(), results.size()); for (List<Integer> childCursorList : elems) { int pos = 0; for (Integer result : results) { if (pos < childCursorList.size() && result.equals(childCursorList.get(pos))) { pos++; } } assertEquals(childCursorList.size(), pos); } RecordCursorResult<Integer> noNextResult = cursor.getNext(); assertThat(noNextResult.hasNext(), is(false)); assertThat(noNextResult.getNoNextReason().isSourceExhausted(), is(true)); }
@Test public void roundRobin() { final List<Integer> expectedResults = Arrays.asList(0, 401, 2, 100, 201, 302, 200, 1, 102); final List<FirableCursor<Integer>> cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 100, 200))), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(401, 201, 1))), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(2, 302, 102))) ); final RecordCursorIterator<Integer> cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); Iterator<Integer> expectedIterator = expectedResults.iterator(); int currentCursor = 0; while (expectedIterator.hasNext()) { cursors.get(currentCursor).fire(); int nextValue = cursor.next(); assertEquals((int)expectedIterator.next(), nextValue); currentCursor = (currentCursor + 1) % cursors.size(); } cursors.forEach(FirableCursor::fireAll); assertThat(cursor.hasNext(), is(false)); assertThat(cursor.getNoNextReason().isSourceExhausted(), is(true)); }
@Test public void concat() { final int childCursorSize = 3; final List<Integer> expectedResults = Arrays.asList(0, 100, 200, 401, 201, 1, 2, 302, 102); final List<FirableCursor<Integer>> cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 100, 200))), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(401, 201, 1))), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(2, 302, 102))) ); final RecordCursorIterator<Integer> cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); Iterator<Integer> expectedIterator = expectedResults.iterator(); for (FirableCursor<Integer> childCursor : cursors) { childCursor.fireAll(); for (int i = 0; i < childCursorSize; i++) { assertEquals((int)expectedIterator.next(), (int)cursor.next()); } assertThat(childCursor.getNext().hasNext(), is(false)); } assertThat(cursor.hasNext(), is(false)); assertThat(cursor.getNoNextReason().isSourceExhausted(), is(true)); }
@Test public void innerLimitReasons() throws ExecutionException, InterruptedException { List<FirableCursor<Integer>> cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 1)).limitRowsTo(1)), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(3, 4))) ); RecordCursorIterator<Integer> cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); cursors.get(0).fireAll(); assertEquals(0, (int)cursor.next()); cursors.get(1).fire(); assertEquals(3, (int)cursor.next()); cursors.get(1).fire(); assertEquals(4, (int)cursor.next()); cursors.get(1).fire(); assertThat(cursor.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursor.getNoNextReason()); cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 1)).limitRowsTo(1)), new FirableCursor<>(RecordCursor.fromList(Arrays.asList(3, 4))) ); cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); cursors.get(1).fire(); assertEquals(3, (int)cursor.next()); cursors.get(0).fireAll(); assertEquals(0, (int)cursor.next()); cursors.get(1).fire(); assertEquals(4, (int)cursor.next()); cursors.get(1).fire(); assertThat(cursor.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursor.getNoNextReason()); cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 1)).limitRowsTo(1)), new FirableCursor<>(RecordCursor.fromList(Collections.singletonList(3))) ); cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); cursors.get(1).fire(); assertEquals(3, (int)cursor.next()); cursors.get(0).fireAll(); cursors.get(1).fireAll(); assertEquals(0, (int)cursor.next()); assertThat(cursor.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursor.getNoNextReason()); cursors = Arrays.asList( new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 1)).limitRowsTo(1)), new FirableCursor<>(new RecordCursorTest.FakeOutOfBandCursor<>(RecordCursor.fromList(Arrays.asList(3, 4)), 1, RecordCursor.NoNextReason.SCAN_LIMIT_REACHED)) ); cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); cursors.get(1).fire(); assertEquals(3, (int)cursor.next()); cursors.get(0).fireAll(); assertEquals(0, (int)cursor.next()); cursors.get(1).fireAll(); assertThat(cursor.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.SCAN_LIMIT_REACHED, cursor.getNoNextReason()); }
@Test public void childCompletesBetweenHasNextAndNext() { final FirableCursor<Integer> cursor1 = new FirableCursor<>(RecordCursor.fromList(Arrays.asList(0, 1))); final FirableCursor<Integer> cursor2 = new FirableCursor<>(RecordCursor.fromList(Arrays.asList(3, 4)).limitRowsTo(1)); List<FirableCursor<Integer>> cursors = Arrays.asList(cursor1, cursor2); RecordCursorIterator<Integer> cursor = UnorderedUnionCursor.create(functionsFromCursors(cursors), null, null).asIterator(); cursor2.fire(); assertEquals(3, (int)cursor.next()); cursor1.fire(); assertThat(cursor.hasNext(), is(true)); cursor2.fire(); assertThat(cursor2.getNext().hasNext(), is(false)); assertEquals(0, (int)cursor.next()); cursor1.fire(); assertEquals(1, (int)cursor.next()); cursor1.fire(); assertThat(cursor.hasNext(), is(false)); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursor.getNoNextReason()); }
@Test public void errorInChild() { CompletableFuture<Integer> future = new CompletableFuture<>(); RecordCursor<Integer> cursor = UnorderedUnionCursor.create(Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation), continuation -> RecordCursor.fromFuture(future) ), null, null); RecordCursorResult<Integer> cursorResult = cursor.getNext(); assertEquals(1, (int)cursorResult.get()); cursorResult = cursor.getNext(); assertEquals(2, (int)cursorResult.get()); 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 = UnorderedUnionCursor.create(Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation).limitRowsTo(1), continuation -> RecordCursor.fromFuture(future) ), null, null); RecordCursorResult<Integer> cursorResult = cursor.getNext(); assertEquals(1, (int)cursorResult.get()); 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(3, 4))); RecordCursor<Integer> cursor = UnorderedUnionCursor.create(Arrays.asList( continuation -> RecordCursor.fromList(Arrays.asList(1, 2), continuation).limitRowsTo(1), continuation -> secondCursor ), null, timer); RecordCursorResult<Integer> cursorResult = cursor.getNext(); assertEquals(1, (int)cursorResult.get()); CompletableFuture<RecordCursorResult<Integer>> cursorResultFuture = cursor.onNext(); assertFalse(cursorResultFuture.isDone()); secondCursor.fire(); cursorResult = cursorResultFuture.get(); assertEquals(3, (int)cursorResult.get()); cursorResultFuture = cursor.onNext(); assertFalse(cursorResultFuture.isDone()); secondCursor.fire(); cursorResult = cursorResultFuture.get(); assertEquals(4, (int)cursorResult.get()); cursorResultFuture = cursor.onNext(); assertFalse(cursorResultFuture.isDone()); secondCursor.fire(); cursorResult = cursorResultFuture.get(); assertFalse(cursorResult.hasNext()); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, cursorResult.getNoNextReason()); assertThat(timer.getCount(FDBStoreTimer.Events.QUERY_INTERSECTION), lessThanOrEqualTo(5)); }