address comment.

Author: hatyo

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

@@ -78,7 +78,7 @@ public class FlatMapPipelinedCursor<T, V> implements RecordCursor<V> {
      * should generally be mediated through one of the {@code synchronized} methods.
      */
     @Nonnull
-    private final Queue<PipelineQueueEntry> pipeline;
+    private final Queue<PipelineQueueEntry<V, T>> pipeline;
     /**
      * The next value to pull from the outer cursor. This value is cleared out by {@link #close()}, so
      * some care should be taken when accessing it. In general, it should be set via one of the {@code synchronized}
@@ -124,7 +124,7 @@ public CompletableFuture<RecordCursorResult<V>> onNext() {
             return CompletableFuture.completedFuture(lastResult);
         }
         return AsyncUtil.whileTrue(this::tryToFillPipeline, getExecutor()).thenApply(vignore -> {
-            PipelineQueueEntry peeked = peekPipeline();
+            PipelineQueueEntry<V, T> peeked = peekPipeline();
             if (peeked == null) {
                 throw new CancellationException("cursor closed while iterating");
             }
@@ -190,12 +190,12 @@ protected CompletableFuture<Boolean> tryToFillPipeline() {
 
             if (!outerNext.isDone()) {
                 // Still waiting for outer future. Check back when it has finished.
-                final PipelineQueueEntry nextEntry = peekPipeline();
+                final PipelineQueueEntry<V, T> nextEntry = peekPipeline();
                 if (nextEntry == null) {
                     return outerNext.thenApply(vignore -> true); // loop back to process outer result
                 } else {
                     // keep looping unless we get something from the next entry's inner cursor or the next cursor is ready
-                    final CompletableFuture<PipelineQueueEntry> innerPipelineFuture = nextEntry.getNextInnerPipelineFuture();
+                    final CompletableFuture<PipelineQueueEntry<V, T>> innerPipelineFuture = nextEntry.getNextInnerPipelineFuture();
                     return CompletableFuture.anyOf(outerNext, innerPipelineFuture).thenApply(vignore ->
                         !innerPipelineFuture.isDone() || innerPipelineFuture.join().doesNotHaveReturnableResult());
                 }
@@ -220,12 +220,12 @@ protected CompletableFuture<Boolean> tryToFillPipeline() {
                 }
                 final RecordCursor<V> innerCursor = innerCursorFunction.apply(outerValue, innerContinuation);
                 outerContinuation = outerResult.getContinuation();
-                addEntryToPipeline(new PipelineQueueEntry(innerCursor, priorOuterContinuation, outerResult, outerCheckValue));
+                addEntryToPipeline(PipelineQueueEntry.newInstanceWithBackgroundComputationOfFirstResult(innerCursor, priorOuterContinuation, outerResult, outerCheckValue));
                 outerNextFuture = null; // done with this future, advance outer cursor next time
                 // keep looping to fill pipeline
             } else { // don't have next, and won't ever with this cursor
                 // Add sentinel to end of pipeline
-                addEntryToPipeline(new PipelineQueueEntry(null, outerContinuation, outerResult, null));
+                addEntryToPipeline(PipelineQueueEntry.newSentinel(outerContinuation, outerResult));
                 outerExhausted = true;
                 // Wait for next entry, as if pipeline were full
                 break;
@@ -239,7 +239,7 @@ protected CompletableFuture<Boolean> tryToFillPipeline() {
         // 4) A concurrent operation cancelled this cursor and so the pipeline is empty
         // The only case where the element should be null is when the cursor has been closed, so return CANCELLED in
         // that case.
-        PipelineQueueEntry peeked = peekPipeline();
+        PipelineQueueEntry<V, T> peeked = peekPipeline();
         if (peeked == null) {
             return ALREADY_CANCELLED;
         }
@@ -263,45 +263,42 @@ private synchronized CompletableFuture<RecordCursorResult<T>> ensureOuterCursorA
         return outerNextFuture;
     }
 
-    private synchronized void addEntryToPipeline(PipelineQueueEntry pipelineQueueEntry) {
+    private synchronized void addEntryToPipeline(PipelineQueueEntry<V, T> pipelineQueueEntry) {
         if (closed) {
             pipelineQueueEntry.close();
         }
-        pipelineQueueEntry.getNextInnerPipelineFuture();
         pipeline.add(pipelineQueueEntry);
     }
 
     private synchronized boolean continueFillingPipeline() {
         return !closed && !outerExhausted && pipeline.size() < pipelineSize;
     }
 
-    private synchronized PipelineQueueEntry peekPipeline() {
+    private synchronized PipelineQueueEntry<V, T> peekPipeline() {
         // Not a lot in this method, but the underlying queue isn't thread safe so
         return pipeline.peek();
     }
 
-    private class PipelineQueueEntry {
+    private static class PipelineQueueEntry<V, T> {
         final RecordCursor<V> innerCursor;
         final RecordCursorContinuation priorOuterContinuation;
         final RecordCursorResult<T> outerResult;
         final byte[] outerCheckValue;
 
         private CompletableFuture<RecordCursorResult<V>> innerFuture;
 
-        public PipelineQueueEntry(RecordCursor<V> innerCursor,
+        private PipelineQueueEntry(RecordCursor<V> innerCursor,
                                   RecordCursorContinuation priorOuterContinuation,
                                   RecordCursorResult<T> outerResult,
                                   byte[] outerCheckValue) {
             this.innerCursor = innerCursor;
             this.priorOuterContinuation = priorOuterContinuation;
             this.outerResult = outerResult;
             this.outerCheckValue = outerCheckValue;
-            // start calculating the next result in the background.
-            setInnerFuture();
         }
 
         @Nonnull
-        public CompletableFuture<PipelineQueueEntry> getNextInnerPipelineFuture() {
+        public CompletableFuture<PipelineQueueEntry<V, T>> getNextInnerPipelineFuture() {
             if (innerFuture == null) {
                 setInnerFuture();
             }
@@ -366,6 +363,30 @@ public RecordCursorResult<V> nextResult() {
         private Continuation<T, V> toContinuation() {
             return new Continuation<>(priorOuterContinuation, outerResult, outerCheckValue, innerFuture.join());
         }
+
+        @Nonnull
+        public static <V, T> PipelineQueueEntry<V, T> newInstance(RecordCursor<V> innerCursor,
+                                                                  RecordCursorContinuation priorOuterContinuation,
+                                                                  RecordCursorResult<T> outerResult,
+                                                                  byte[] outerCheckValue) {
+            return new PipelineQueueEntry<>(innerCursor, priorOuterContinuation, outerResult, outerCheckValue);
+        }
+
+        @Nonnull
+        public static <V, T> PipelineQueueEntry<V, T> newInstanceWithBackgroundComputationOfFirstResult(RecordCursor<V> innerCursor,
+                                                                                                        RecordCursorContinuation priorOuterContinuation,
+                                                                                                        RecordCursorResult<T> outerResult,
+                                                                                                        byte[] outerCheckValue) {
+            final var result = newInstance(innerCursor, priorOuterContinuation, outerResult, outerCheckValue);
+            result.setInnerFuture();
+            return result;
+        }
+
+        @Nonnull
+        public static <V, T> PipelineQueueEntry<V, T> newSentinel(RecordCursorContinuation priorOuterContinuation,
+                                                                  RecordCursorResult<T> outerResult) {
+            return new PipelineQueueEntry<>(null, priorOuterContinuation, outerResult, null);
+        }
     }
 
     private static class Continuation<T, V> implements RecordCursorContinuation {