Start calculating inner's first item concurrently.

- this should improve the performance of the pipeline now that it schedule
  up to N-calculations of the inner's first cursor item, where N is the size
  of the pipeline.

Author: hatyo

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

@@ -267,6 +267,7 @@ private synchronized void addEntryToPipeline(PipelineQueueEntry pipelineQueueEnt
         if (closed) {
             pipelineQueueEntry.close();
         }
+        pipelineQueueEntry.getNextInnerPipelineFuture();
         pipeline.add(pipelineQueueEntry);
     }
 
@@ -295,20 +296,26 @@ public PipelineQueueEntry(RecordCursor<V> innerCursor,
             this.priorOuterContinuation = priorOuterContinuation;
             this.outerResult = outerResult;
             this.outerCheckValue = outerCheckValue;
+            // start calculating the next result in the background.
+            setInnerFuture();
         }
 
         @Nonnull
         public CompletableFuture<PipelineQueueEntry> getNextInnerPipelineFuture() {
             if (innerFuture == null) {
-                if (innerCursor == null) {
-                    innerFuture = CompletableFuture.completedFuture(RecordCursorResult.exhausted());
-                } else {
-                    innerFuture = innerCursor.onNext();
-                }
+                setInnerFuture();
             }
             return innerFuture.thenApply(vignore -> this);
         }
 
+        private void setInnerFuture() {
+            if (innerCursor == null) {
+                innerFuture = CompletableFuture.completedFuture(RecordCursorResult.exhausted());
+            } else {
+                innerFuture = innerCursor.onNext();
+            }
+        }
+
         public boolean doesNotHaveReturnableResult() {
             if (innerCursor == null ||       // Hit sentinel, so we have a returnable result
                     innerFuture == null ||   // Inner future hasn't been started yet.