Defer execution 2024

src/main/java/graphql/ExecutionInput.java

@@ -383,4 +383,4 @@ public ExecutionInput build() {
             return new ExecutionInput(this);
         }
     }
-}
+}

src/main/java/graphql/ExecutionResultImpl.java

@@ -40,6 +40,10 @@ public ExecutionResultImpl(ExecutionResultImpl other) {
         this(other.dataPresent, other.data, other.errors, other.extensions);
     }
 
+    public <T extends Builder<T>> ExecutionResultImpl(Builder<T> builder) {
+        this(builder.dataPresent, builder.data, builder.errors, builder.extensions);
+    }
+
     private ExecutionResultImpl(boolean dataPresent, Object data, List<? extends GraphQLError> errors, Map<Object, Object> extensions) {
         this.dataPresent = dataPresent;
         this.data = data;
@@ -103,61 +107,61 @@ public String toString() {
                 '}';
     }
 
-    public static Builder newExecutionResult() {
-        return new Builder();
+    public static <T extends Builder<T>> Builder<T> newExecutionResult() {
+        return new Builder<>();
     }
 
-    public static class Builder implements ExecutionResult.Builder<Builder> {
+    public static class Builder<T extends Builder<T>> implements ExecutionResult.Builder<T> {
         private boolean dataPresent;
         private Object data;
         private List<GraphQLError> errors = new ArrayList<>();
         private Map<Object, Object> extensions;
 
         @Override
-        public Builder from(ExecutionResult executionResult) {
+        public T from(ExecutionResult executionResult) {
             dataPresent = executionResult.isDataPresent();
             data = executionResult.getData();
             errors = new ArrayList<>(executionResult.getErrors());
             extensions = executionResult.getExtensions();
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder data(Object data) {
+        public T data(Object data) {
             dataPresent = true;
             this.data = data;
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder errors(List<GraphQLError> errors) {
+        public T errors(List<GraphQLError> errors) {
             this.errors = errors;
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder addErrors(List<GraphQLError> errors) {
+        public T addErrors(List<GraphQLError> errors) {
             this.errors.addAll(errors);
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder addError(GraphQLError error) {
+        public T addError(GraphQLError error) {
             this.errors.add(error);
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder extensions(Map<Object, Object> extensions) {
+        public T extensions(Map<Object, Object> extensions) {
             this.extensions = extensions;
-            return this;
+            return (T) this;
         }
 
         @Override
-        public Builder addExtension(String key, Object value) {
+        public T addExtension(String key, Object value) {
             this.extensions = (this.extensions == null ? new LinkedHashMap<>() : this.extensions);
             this.extensions.put(key, value);
-            return this;
+            return (T) this;
         }
 
         @Override

src/main/java/graphql/ExperimentalApi.java

@@ -20,4 +20,8 @@
 @Target(value = {CONSTRUCTOR, METHOD, TYPE, FIELD})
 @Documented
 public @interface ExperimentalApi {
+    /**
+     * The key that should be associated with a boolean value which indicates whether @defer and @stream behaviour is enabled for this execution.
+     */
+    String ENABLE_INCREMENTAL_SUPPORT  = "ENABLE_INCREMENTAL_SUPPORT";
 }

src/main/java/graphql/execution/AsyncExecutionStrategy.java

@@ -2,6 +2,7 @@
 
 import graphql.ExecutionResult;
 import graphql.PublicApi;
+import graphql.execution.incremental.DeferredExecutionSupport;
 import graphql.execution.instrumentation.ExecutionStrategyInstrumentationContext;
 import graphql.execution.instrumentation.Instrumentation;
 import graphql.execution.instrumentation.parameters.InstrumentationExecutionStrategyParameters;
@@ -10,7 +11,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.function.BiConsumer;
 
-
 /**
  * The standard graphql execution strategy that runs fields asynchronously non-blocking.
  */
@@ -36,19 +36,24 @@ public AsyncExecutionStrategy(DataFetcherExceptionHandler exceptionHandler) {
     @Override
     @SuppressWarnings("FutureReturnValueIgnored")
     public CompletableFuture<ExecutionResult> execute(ExecutionContext executionContext, ExecutionStrategyParameters parameters) throws NonNullableFieldWasNullException {
-
         Instrumentation instrumentation = executionContext.getInstrumentation();
         InstrumentationExecutionStrategyParameters instrumentationParameters = new InstrumentationExecutionStrategyParameters(executionContext, parameters);
 
         ExecutionStrategyInstrumentationContext executionStrategyCtx = ExecutionStrategyInstrumentationContext.nonNullCtx(instrumentation.beginExecutionStrategy(instrumentationParameters, executionContext.getInstrumentationState()));
 
-        List<String> fieldNames = parameters.getFields().getKeys();
-        Async.CombinedBuilder<FieldValueInfo> futures = getAsyncFieldValueInfo(executionContext, parameters);
+        MergedSelectionSet fields = parameters.getFields();
+        List<String> fieldNames = fields.getKeys();
+
+        DeferredExecutionSupport deferredExecutionSupport = createDeferredExecutionSupport(executionContext, parameters);
+        Async.CombinedBuilder<FieldValueInfo> futures = getAsyncFieldValueInfo(executionContext, parameters, deferredExecutionSupport);
+
         CompletableFuture<ExecutionResult> overallResult = new CompletableFuture<>();
         executionStrategyCtx.onDispatched(overallResult);
 
         futures.await().whenComplete((completeValueInfos, throwable) -> {
-            BiConsumer<List<Object>, Throwable> handleResultsConsumer = handleResults(executionContext, fieldNames, overallResult);
+            List<String> fieldsExecutedOnInitialResult = deferredExecutionSupport.getNonDeferredFieldNames(fieldNames);
+
+            BiConsumer<List<Object>, Throwable> handleResultsConsumer = handleResults(executionContext, fieldsExecutedOnInitialResult, overallResult);
             if (throwable != null) {
                 handleResultsConsumer.accept(null, throwable.getCause());
                 return;

src/main/java/graphql/execution/Execution.java

@@ -4,15 +4,19 @@
 import graphql.ExecutionInput;
 import graphql.ExecutionResult;
 import graphql.ExecutionResultImpl;
+import graphql.ExperimentalApi;
 import graphql.GraphQLContext;
 import graphql.GraphQLError;
 import graphql.Internal;
+import graphql.execution.incremental.IncrementalCallState;
 import graphql.execution.instrumentation.Instrumentation;
 import graphql.execution.instrumentation.InstrumentationContext;
 import graphql.execution.instrumentation.InstrumentationState;
 import graphql.execution.instrumentation.parameters.InstrumentationExecuteOperationParameters;
 import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters;
 import graphql.extensions.ExtensionsBuilder;
+import graphql.incremental.DelayedIncrementalPartialResult;
+import graphql.incremental.IncrementalExecutionResultImpl;
 import graphql.language.Document;
 import graphql.language.FragmentDefinition;
 import graphql.language.NodeUtil;
@@ -21,10 +25,12 @@
 import graphql.schema.GraphQLObjectType;
 import graphql.schema.GraphQLSchema;
 import graphql.schema.impl.SchemaUtil;
+import org.reactivestreams.Publisher;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 import static graphql.execution.ExecutionContextBuilder.newExecutionContextBuilder;
@@ -133,7 +139,13 @@ private CompletableFuture<ExecutionResult> executeOperation(ExecutionContext exe
                 .graphQLContext(graphQLContext)
                 .build();
 
-        MergedSelectionSet fields = fieldCollector.collectFields(collectorParameters, operationDefinition.getSelectionSet());
+        MergedSelectionSet fields = fieldCollector.collectFields(
+                collectorParameters,
+                operationDefinition.getSelectionSet(),
+                Optional.ofNullable(executionContext.getGraphQLContext())
+                        .map(graphqlContext -> (Boolean) graphqlContext.get(ExperimentalApi.ENABLE_INCREMENTAL_SUPPORT))
+                        .orElse(false)
+        );
 
         ResultPath path = ResultPath.rootPath();
         ExecutionStepInfo executionStepInfo = newExecutionStepInfo().type(operationRootType).path(path).build();
@@ -170,7 +182,31 @@ private CompletableFuture<ExecutionResult> executeOperation(ExecutionContext exe
         result = result.thenApply(er -> mergeExtensionsBuilderIfPresent(er, graphQLContext));
 
         result = result.whenComplete(executeOperationCtx::onCompleted);
-        return result;
+
+        return incrementalSupport(executionContext, result);
+    }
+
+    /*
+     * Adds the deferred publisher if it's needed at the end of the query.  This is also a good time for the deferred code to start running
+     */
+    private CompletableFuture<ExecutionResult> incrementalSupport(ExecutionContext executionContext, CompletableFuture<ExecutionResult> result) {
+        return result.thenApply(er -> {
+            IncrementalCallState incrementalCallState = executionContext.getIncrementalCallState();
+            if (incrementalCallState.getIncrementalCallsDetected()) {
+                // we start the rest of the query now to maximize throughput.  We have the initial important results,
+                // and now we can start the rest of the calls as early as possible (even before someone subscribes)
+                Publisher<DelayedIncrementalPartialResult> publisher = incrementalCallState.startDeferredCalls();
+
+                return IncrementalExecutionResultImpl.fromExecutionResult(er)
+                        // "hasNext" can, in theory, be "false" when all the incremental items are delivered in the
+                        // first response payload. However, the current implementation will never result in this.
+                        // The behaviour might change if we decide to make optimizations in the future.
+                        .hasNext(true)
+                        .incrementalItemPublisher(publisher)
+                        .build();
+            }
+            return er;
+        });
     }
 
     private void addExtensionsBuilderNotPresent(GraphQLContext graphQLContext) {

src/main/java/graphql/execution/ExecutionContext.java

@@ -8,6 +8,7 @@
 import graphql.GraphQLError;
 import graphql.PublicApi;
 import graphql.collect.ImmutableKit;
+import graphql.execution.incremental.IncrementalCallState;
 import graphql.execution.instrumentation.Instrumentation;
 import graphql.execution.instrumentation.InstrumentationState;
 import graphql.language.Document;
@@ -53,6 +54,7 @@ public class ExecutionContext {
     private final Set<ResultPath> errorPaths = new HashSet<>();
     private final DataLoaderRegistry dataLoaderRegistry;
     private final Locale locale;
+    private final IncrementalCallState incrementalCallState = new IncrementalCallState();
     private final ValueUnboxer valueUnboxer;
     private final ExecutionInput executionInput;
     private final Supplier<ExecutableNormalizedOperation> queryTree;
@@ -255,6 +257,10 @@ public ExecutionStrategy getSubscriptionStrategy() {
         return subscriptionStrategy;
     }
 
+    public IncrementalCallState getIncrementalCallState() {
+        return incrementalCallState;
+    }
+
     public ExecutionStrategy getStrategy(OperationDefinition.Operation operation) {
         if (operation == OperationDefinition.Operation.MUTATION) {
             return getMutationStrategy();