introduce legacy incremental executor (#4546)

Motivation: to ease user base upgrade path. The legacy incremental executor uses the legacy format which is not expected to be included within the spec and clashes with the new format.

Note: the legacy incremental executor is a branching executor, it uses the legacy format that duplicates fields within the response as well as the underlying legacy behavior in which those fields are re-executed, with potentially conflicting field values. In particular, nulls may bubble up in one branch and not another. The transformer approach from #4319 keeps the legacy format without re-executing because it transforms the non-branching incremental executor response.

[It was previously thought not possible to release a legacy incremental executor using the same underlying machinery as the new one, see #4319, but I cannot understand the explanation I previously gave to that effect.]

Author: yaacovCR

src/execution/__tests__/executor-test.ts

@@ -42,12 +42,14 @@ import {
 } from '../execute.js';
 import type { ExecutionResult } from '../Executor.js';
 import { collectSubfields, getStreamUsage } from '../Executor.js';
+import { legacyExecuteIncrementally } from '../legacyIncremental/legacyExecuteIncrementally.js';
 
 function execute(args: ExecutionArgs): PromiseOrValue<ExecutionResult> {
   return expectEqualPromisesOrValues([
     executeThrowingOnIncremental(args),
     executeIgnoringIncremental(args),
     experimentalExecuteIncrementally(args),
+    legacyExecuteIncrementally(args),
   ]) as PromiseOrValue<ExecutionResult>;
 }
 
@@ -56,6 +58,7 @@ function executeSync(args: ExecutionArgs): ExecutionResult {
     executeSyncWrappingThrowingOnIncremental(args),
     executeIgnoringIncremental(args),
     experimentalExecuteIncrementally(args),
+    legacyExecuteIncrementally(args),
   ]) as ExecutionResult;
 }
 

src/execution/incremental/IncrementalExecutor.ts

@@ -273,6 +273,12 @@ export class IncrementalExecutor<
     this.streams = [];
   }
 
+  createSubExecutor(
+    deferUsageSet?: DeferUsageSet,
+  ): IncrementalExecutor<TExperimental> {
+    return new IncrementalExecutor(this.validatedExecutionArgs, deferUsageSet);
+  }
+
   override cancel(reason?: unknown): void {
     super.cancel(reason);
     for (const task of this.tasks) {
@@ -442,10 +448,7 @@ export class IncrementalExecutor<
     for (const [deferUsageSet, groupedFieldSet] of newGroupedFieldSets) {
       const deliveryGroups = getDeliveryGroups(deferUsageSet, deliveryGroupMap);
 
-      const executor = new IncrementalExecutor(
-        this.validatedExecutionArgs,
-        deferUsageSet,
-      );
+      const executor = this.createSubExecutor(deferUsageSet);
 
       const executionGroup: ExecutionGroup = {
         groups: deliveryGroups,
@@ -708,7 +711,7 @@ export class IncrementalExecutor<
 
           const itemPath = addPath(streamPath, index, undefined);
 
-          const executor = new IncrementalExecutor(this.validatedExecutionArgs);
+          const executor = this.createSubExecutor();
 
           let streamItemResult = executor.completeStreamItem(
             itemPath,

src/execution/incremental/__tests__/defer-test.ts

@@ -2132,7 +2132,43 @@ describe('Execute: defer directive', () => {
     ]);
   });
 
-  it('Cancels deferred fields when initial result exhibits null bubbling', async () => {
+  it('Cancels deferred fields when initial result exhibits null bubbling cancelling the defer', async () => {
+    const document = parse(`
+      query {
+        hero {
+          nonNullName
+          ... @defer {
+            name
+          }
+        }
+      }
+    `);
+    const result = await complete(
+      document,
+      {
+        hero: {
+          ...hero,
+          nonNullName: () => null,
+        },
+      },
+      true,
+    );
+    expectJSON(result).toDeepEqual({
+      data: {
+        hero: null,
+      },
+      errors: [
+        {
+          message:
+            'Cannot return null for non-nullable field Hero.nonNullName.',
+          locations: [{ line: 4, column: 11 }],
+          path: ['hero', 'nonNullName'],
+        },
+      ],
+    });
+  });
+
+  it('Cancels deferred fields when initial result exhibits null bubbling cancelling new fields', async () => {
     const document = parse(`
       query {
         hero {

src/execution/legacyIncremental/BranchingIncrementalExecutor.ts

@@ -0,0 +1,169 @@
+import { AccumulatorMap } from '../../jsutils/AccumulatorMap.js';
+import { getBySet } from '../../jsutils/getBySet.js';
+import { invariant } from '../../jsutils/invariant.js';
+import { isSameSet } from '../../jsutils/isSameSet.js';
+import { memoize1 } from '../../jsutils/memoize1.js';
+import { memoize2 } from '../../jsutils/memoize2.js';
+import type { ObjMap } from '../../jsutils/ObjMap.js';
+
+import type { GraphQLError } from '../../error/GraphQLError.js';
+
+import type {
+  DeferUsage,
+  FieldDetails,
+  GroupedFieldSet,
+} from '../collectFields.js';
+import type { ExecutionResult } from '../Executor.js';
+import type {
+  DeferUsageSet,
+  ExecutionPlan,
+} from '../incremental/buildExecutionPlan.js';
+import { IncrementalExecutor } from '../incremental/IncrementalExecutor.js';
+
+import { BranchingIncrementalPublisher } from './BranchingIncrementalPublisher.js';
+
+export interface ExperimentalIncrementalExecutionResults {
+  initialResult: InitialIncrementalExecutionResult;
+  subsequentResults: AsyncGenerator<
+    SubsequentIncrementalExecutionResult,
+    void,
+    void
+  >;
+}
+
+export interface InitialIncrementalExecutionResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> extends ExecutionResult<TData, TExtensions> {
+  data: TData;
+  hasNext: true;
+  extensions?: TExtensions;
+}
+
+export interface SubsequentIncrementalExecutionResult<
+  TData = unknown,
+  TExtensions = ObjMap<unknown>,
+> {
+  incremental?: ReadonlyArray<IncrementalResult<TData, TExtensions>>;
+  hasNext: boolean;
+  extensions?: TExtensions;
+}
+
+export type IncrementalResult<TData = unknown, TExtensions = ObjMap<unknown>> =
+  | IncrementalDeferResult<TData, TExtensions>
+  | IncrementalStreamResult<TData, TExtensions>;
+
+export interface IncrementalDeferResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> extends ExecutionResult<TData, TExtensions> {
+  path: ReadonlyArray<string | number>;
+  label?: string;
+}
+
+export interface IncrementalStreamResult<
+  TData = ReadonlyArray<unknown>,
+  TExtensions = ObjMap<unknown>,
+> {
+  errors?: ReadonlyArray<GraphQLError>;
+  items: TData | null;
+  path: ReadonlyArray<string | number>;
+  label?: string;
+  extensions?: TExtensions;
+}
+
+const buildBranchingExecutionPlanFromInitial = memoize1(
+  (groupedFieldSet: GroupedFieldSet) =>
+    buildBranchingExecutionPlan(groupedFieldSet),
+);
+
+const buildBranchingExecutionPlanFromDeferred = memoize2(
+  (groupedFieldSet: GroupedFieldSet, deferUsageSet: DeferUsageSet) =>
+    buildBranchingExecutionPlan(groupedFieldSet, deferUsageSet),
+);
+
+/** @internal */
+export class BranchingIncrementalExecutor extends IncrementalExecutor<ExperimentalIncrementalExecutionResults> {
+  override createSubExecutor(
+    deferUsageSet?: DeferUsageSet,
+  ): IncrementalExecutor<ExperimentalIncrementalExecutionResults> {
+    return new BranchingIncrementalExecutor(
+      this.validatedExecutionArgs,
+      deferUsageSet,
+    );
+  }
+
+  override buildResponse(
+    data: ObjMap<unknown> | null,
+  ): ExecutionResult | ExperimentalIncrementalExecutionResults {
+    const errors = this.collectedErrors.errors;
+    const work = this.getIncrementalWork();
+    const { tasks, streams } = work;
+    if (tasks?.length === 0 && streams?.length === 0) {
+      return errors.length ? { errors, data } : { data };
+    }
+
+    invariant(data !== null);
+    const incrementalPublisher = new BranchingIncrementalPublisher();
+    return incrementalPublisher.buildResponse(
+      data,
+      errors,
+      work,
+      this.validatedExecutionArgs.externalAbortSignal,
+    );
+  }
+
+  override buildRootExecutionPlan(
+    originalGroupedFieldSet: GroupedFieldSet,
+  ): ExecutionPlan {
+    return buildBranchingExecutionPlanFromInitial(originalGroupedFieldSet);
+  }
+
+  override buildSubExecutionPlan(
+    originalGroupedFieldSet: GroupedFieldSet,
+  ): ExecutionPlan {
+    return this.deferUsageSet === undefined
+      ? buildBranchingExecutionPlanFromInitial(originalGroupedFieldSet)
+      : buildBranchingExecutionPlanFromDeferred(
+          originalGroupedFieldSet,
+          this.deferUsageSet,
+        );
+  }
+}
+
+function buildBranchingExecutionPlan(
+  originalGroupedFieldSet: GroupedFieldSet,
+  parentDeferUsages: DeferUsageSet = new Set<DeferUsage>(),
+): ExecutionPlan {
+  const groupedFieldSet = new AccumulatorMap<string, FieldDetails>();
+
+  const newGroupedFieldSets = new Map<
+    DeferUsageSet,
+    AccumulatorMap<string, FieldDetails>
+  >();
+
+  for (const [responseKey, fieldGroup] of originalGroupedFieldSet) {
+    for (const fieldDetails of fieldGroup) {
+      const deferUsage = fieldDetails.deferUsage;
+      const deferUsageSet =
+        deferUsage === undefined
+          ? new Set<DeferUsage>()
+          : new Set([deferUsage]);
+      if (isSameSet(parentDeferUsages, deferUsageSet)) {
+        groupedFieldSet.add(responseKey, fieldDetails);
+      } else {
+        let newGroupedFieldSet = getBySet(newGroupedFieldSets, deferUsageSet);
+        if (newGroupedFieldSet === undefined) {
+          newGroupedFieldSet = new AccumulatorMap();
+          newGroupedFieldSets.set(deferUsageSet, newGroupedFieldSet);
+        }
+        newGroupedFieldSet.add(responseKey, fieldDetails);
+      }
+    }
+  }
+
+  return {
+    groupedFieldSet,
+    newGroupedFieldSets,
+  };
+}