EXISTS queries now honor continuation from previous runs to avoid infinite loops (#3221)

The `RecordQueryFirstOrDefaultPlan` constructs a special cursor which
has cardinality at most one. The continuation it returns back to the
user is not based on the inner cursor but only on whether the cursor has
returned any data. Previously, we were taking the `FutureCursor`'s
continuation (which is always `\x00`) and giving it back to the `inner`
plan, which would generally result in the plan resuming from the
beginning. That is not the appropriate way to interpret the
continuation, so this updates the logic to instead skip creating the
inner cursor if we get a non-beginning continuation, which is in line
with what the `FutureCursor`'s continuation means.

This fixes #3219.

Author: alecgrieser

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/RecordCursor.java

@@ -59,6 +59,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
@@ -884,24 +885,57 @@ static <T> RecordCursor<T> fromFuture(@Nonnull CompletableFuture<T> future) {
 
     @Nonnull
     static <T> RecordCursor<T> fromFuture(@Nonnull Executor executor, @Nonnull CompletableFuture<T> future) {
-        return new FutureCursor<>(executor, future);
+        return fromFuture(executor, future, null);
     }
 
     /**
      * Get a new cursor that has the contents of the given future as its only record.
      * If the continuation is nonnull, return an empty cursor since a {@link FutureCursor} returns only a single record.
+     * Note that in general, this can lead to futures being started that are not properly waited on,
+     * so it is generally advised that the user use {@link #fromFuture(Executor, Supplier, byte[])} instead
+     * of this one. See also the warning on {@link #fromFuture(Executor, Supplier, byte[])} for a word
+     * about if the work powering the underlying future has more complicated state that should be
+     * included in the cursor's continuation.
+     *
      * @param executor an executor for executing the future
      * @param future a future that completes to the only element of the cursor
      * @param continuation a continuation from a future cursor to determine whether the cursor has a single element or no elements
      * @param <T> the result type of the future
      * @return a new cursor producing the contents of {@code future} if the continuation is nonnull and an empty cursor otherwise
+     * @see #fromFuture(Executor, Supplier, byte[])
      */
     @Nonnull
     static <T> RecordCursor<T> fromFuture(@Nonnull Executor executor, @Nonnull CompletableFuture<T> future, @Nullable byte[] continuation) {
+        return fromFuture(executor, () -> future, continuation);
+    }
+
+    /**
+     * Get a new cursor that has the contents of the lazily-evaluated future as its only record.
+     * If the continuation is nonnull, return an empty cursor since a {@link FutureCursor} returns only a single record.
+     * In that case, the {@code futureSupplier} will not be evaluated to avoid queuing up asynchronous work
+     * that is not guaranteed to have anyone wait on. For that reason, this should generally be preferred
+     * to {@link #fromFuture(Executor, CompletableFuture, byte[])}.
+     *
+     * <p>
+     * <em>Warning:</em> Using a {@link FutureCursor} here may lead to incorrect results if the underlying
+     * future is backed by something that has some kind of state that should be included in the continuation.
+     * For example, if the future is backed by a cursor that may stop for an out-of-band {@link NoNextReason},
+     * it may be desirable to include the progress included in the underlying cursor in the final continuation.
+     * That is not done by this class, so something more bespoke may be required.
+     * </p>
+     *
+     * @param executor an executor for executing the future
+     * @param futureSupplier a supplier of a future that completes to the only element of the cursor
+     * @param continuation a continuation from a future cursor to determine whether the cursor has a single element or no elements
+     * @param <T> the result type of the future
+     * @return a new cursor producing the contents of {@code future} if the continuation is nonnull and an empty cursor otherwise
+     */
+    @Nonnull
+    static <T> RecordCursor<T> fromFuture(@Nonnull Executor executor, @Nonnull Supplier<CompletableFuture<T>> futureSupplier, @Nullable byte[] continuation) {
         if (continuation == null) {
-            return new FutureCursor<>(executor, future);
+            return new FutureCursor<>(executor, futureSupplier.get());
         } else {
-            return RecordCursor.empty();
+            return RecordCursor.empty(executor);
         }
     }
 

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/FutureCursor.java

@@ -50,6 +50,15 @@ public class FutureCursor<T> implements RecordCursor<T> {
     @Nonnull
     private static final RecordCursorContinuation notDoneContinuation = ByteArrayContinuation.fromNullable(new byte[] {0});
 
+    /**
+     * Internal constructor. Users should generally call
+     * {@link RecordCursor#fromFuture(Executor, java.util.function.Supplier, byte[])}
+     * in order to ensure that continuations from this cursor are properly handled.
+     *
+     * @param executor the executor to associate with this cursor
+     * @param future a future that when completed will provide the single element returned by this cursor
+     */
+    @API(API.Status.INTERNAL)
     public FutureCursor(@Nonnull Executor executor, @Nonnull CompletableFuture<T> future) {
         this.executor = executor;
         this.future = future;

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryFirstOrDefaultPlan.java

@@ -28,12 +28,10 @@
 import com.apple.foundationdb.record.PlanHashable;
 import com.apple.foundationdb.record.PlanSerializationContext;
 import com.apple.foundationdb.record.RecordCursor;
-import com.apple.foundationdb.record.cursors.FutureCursor;
 import com.apple.foundationdb.record.planprotos.PRecordQueryFirstOrDefaultPlan;
 import com.apple.foundationdb.record.planprotos.PRecordQueryPlan;
 import com.apple.foundationdb.record.provider.common.StoreTimer;
 import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
-import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.apple.foundationdb.record.query.plan.cascades.AliasMap;
 import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier;
 import com.apple.foundationdb.record.query.plan.cascades.Memoizer;
@@ -47,6 +45,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.values.DerivedValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
 import com.apple.foundationdb.record.query.plan.cascades.values.translation.TranslationMap;
+import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.google.auto.service.AutoService;
 import com.google.common.base.Verify;
 import com.google.common.collect.ImmutableList;
@@ -96,10 +95,17 @@ public <M extends Message> RecordCursor<QueryResult> executePlan(@Nonnull final
                                                                      @Nonnull final EvaluationContext context,
                                                                      @Nullable final byte[] continuation,
                                                                      @Nonnull final ExecuteProperties executeProperties) {
-        return new FutureCursor<>(store.getExecutor(),
-                getChild().executePlan(store, context, continuation, executeProperties).first()
-                        .thenApply(resultOptional ->
-                                resultOptional.orElseGet(() -> QueryResult.ofComputed(onEmptyResultValue.eval(store, context)))));
+        // Note that a null child continuation is always handed to the child cursor below.
+        // This is because the returned FutureCursor only ever returns a single value, and so if
+        // if that lambda is called, it indicates that the original continuation is null, and we
+        // are starting the plan from the beginning. That this doesn't handle the inner cursor
+        // halting with an out-of-band no-next-reasons, which currently is treated the same way
+        // as the inner cursor being empty.
+        // See: https://github.com/FoundationDB/fdb-record-layer/issues/3220
+        return RecordCursor.fromFuture(store.getExecutor(),
+                () -> getChild().executePlan(store, context, null, executeProperties).first()
+                        .thenApply(resultOptional -> resultOptional.orElseGet(() -> QueryResult.ofComputed(onEmptyResultValue.eval(store, context)))),
+                continuation);
     }
 
     @Override

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/RecordCursorTest.java

@@ -24,7 +24,6 @@
 import com.apple.foundationdb.async.MoreAsyncUtil;
 import com.apple.foundationdb.record.cursors.FilterCursor;
 import com.apple.foundationdb.record.cursors.FirableCursor;
-import com.apple.foundationdb.record.cursors.FutureCursor;
 import com.apple.foundationdb.record.cursors.LazyCursor;
 import com.apple.foundationdb.record.cursors.MapResultCursor;
 import com.apple.foundationdb.record.cursors.RowLimitedCursor;
@@ -63,6 +62,7 @@
 import java.util.TimerTask;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -81,11 +81,13 @@
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.oneOf;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -1313,7 +1315,7 @@ private static RecordCursor<Integer> mapPipelinedCursorToClose(int iteration, Co
     private static RecordCursor<String> singletonFlatMapPipelinedCursorToClose(int iteration, CompletableFuture<Void> signal) {
         return RecordCursor.flatMapPipelined(
                 outerContinuation -> RecordCursor.fromList(EXECUTOR, IntStream.range(0, iteration % 199).boxed().collect(Collectors.toList()), outerContinuation),
-                (outerValue, innerContinuation) -> new FutureCursor<>(EXECUTOR, signal.thenApplyAsync(ignore -> String.valueOf(outerValue), EXECUTOR)),
+                (outerValue, innerContinuation) -> RecordCursor.fromFuture(EXECUTOR, signal.thenApplyAsync(ignore -> String.valueOf(outerValue), EXECUTOR), innerContinuation),
                 null,
                 null,
                 iteration % 19 + 2
@@ -1393,4 +1395,113 @@ void pipelinedCursorAfterClosing(@Nonnull BiFunction<Integer, CompletableFuture<
             assertThat(executionException.getCause(), Matchers.instanceOf(CancellationException.class));
         }
     }
+
+    @Test
+    void futureCursorTest() {
+        CompletableFuture<Integer> future = new CompletableFuture<>();
+        final RecordCursorContinuation continuation;
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, future, null)) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+            future.complete(42);
+            RecordCursorResult<Integer> result = fromFuture.getNext();
+            assertTrue(result.hasNext(), "first result from future should have value");
+            assertEquals(42, result.get());
+            assertFalse(result.getContinuation().isEnd());
+            continuation = result.getContinuation();
+
+            result = fromFuture.getNext();
+            assertFalse(result.hasNext(), "second result from future should have value");
+            assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason());
+            assertTrue(result.getContinuation().isEnd());
+        }
+
+        CompletableFuture<Integer> secondFuture = new CompletableFuture<>();
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, secondFuture, continuation.toBytes())) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+
+            RecordCursorResult<Integer> result = fromFuture.getNext();
+            assertFalse(result.hasNext(), "future should not have value when resuming from a continuation");
+            assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason());
+            assertTrue(result.getContinuation().isEnd());
+        }
+    }
+
+    @Test
+    void futureCursorFromSupplierTest() {
+        CompletableFuture<Integer> future = new CompletableFuture<>();
+        final RecordCursorContinuation continuation;
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, () -> future, null)) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+            future.complete(1066);
+            RecordCursorResult<Integer> result = fromFuture.getNext();
+            assertTrue(result.hasNext(), "first result from future should have value");
+            assertEquals(1066, result.get());
+            assertFalse(result.getContinuation().isEnd());
+            continuation = result.getContinuation();
+
+            result = fromFuture.getNext();
+            assertFalse(result.hasNext(), "second result from future should have value");
+            assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason());
+            assertTrue(result.getContinuation().isEnd());
+        }
+
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, () -> fail("should not be run"), continuation.toBytes())) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+
+            RecordCursorResult<Integer> result = fromFuture.getNext();
+            assertFalse(result.hasNext(), "future should not have value when resuming from a continuation");
+            assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason());
+            assertTrue(result.getContinuation().isEnd());
+        }
+    }
+
+    @Test
+    void futureCursorCompletesWhenUnderlyingCompletes() throws Exception {
+        final CompletableFuture<Integer> future = new CompletableFuture<>();
+        final RecordCursorContinuation continuation;
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, future, null)) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+            CompletableFuture<RecordCursorResult<Integer>> resultFuture = fromFuture.onNext();
+            assertFalse(resultFuture.isDone(), "result should not be done until underlying completes");
+            future.complete(1819);
+
+            // The previous result future should complete quickly. The timeout here is
+            // a bit of a hedge, but as currently written, the resultFuture should actually complete
+            // on this thread, and so the resultFuture should already be completed
+            // by now. But that's not necessarily part of the contract of the cursor, so
+            // instead, just assert here that we complete in a reasonable amount of time
+            RecordCursorResult<Integer> result = resultFuture.get(1, TimeUnit.SECONDS);
+            assertTrue(result.hasNext());
+            assertEquals(1819, result.get());
+            continuation = result.getContinuation();
+            assertFalse(continuation.isEnd());
+            assertFalse(continuation.toByteString().isEmpty());
+            assertThat(continuation.toBytes(), notNullValue());
+        }
+
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, () -> fail("should not run"), continuation.toBytes())) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+            CompletableFuture<RecordCursorResult<Integer>> resultFuture = fromFuture.onNext();
+            assertTrue(resultFuture.isDone(), "empty result should not need to wait");
+            RecordCursorResult<Integer> result = resultFuture.get();
+            assertFalse(result.hasNext());
+            assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason());
+            assertTrue(result.getContinuation().isEnd());
+        }
+    }
+
+    @Test
+    void futureCursorPropagatesError() {
+        final CompletableFuture<Integer> future = new CompletableFuture<>();
+        try (RecordCursor<Integer> fromFuture = RecordCursor.fromFuture(EXECUTOR, future, null)) {
+            assertSame(EXECUTOR, fromFuture.getExecutor());
+            CompletableFuture<RecordCursorResult<Integer>> resultFuture = fromFuture.onNext();
+            assertFalse(resultFuture.isDone(), "result should not be done until underlying completes");
+
+            final RecordCoreException error = new RecordCoreException("test error");
+            future.completeExceptionally(error);
+            CompletionException futureError = assertThrows(CompletionException.class, future::join);
+            assertSame(error, futureError.getCause(), "result should complete exception while propagating the cause");
+        }
+    }
 }