simplify CollectFields for @defer and @stream

Replicates graphql/graphql-js@2aedf25

Author: Cito

docs/conf.py

@@ -138,79 +138,77 @@
 }
 
 # ignore the following undocumented or internal references:
-ignore_references = set(
-    [
-        "GNT",
-        "GT",
-        "KT",
-        "T",
-        "VT",
-        "TContext",
-        "Enum",
-        "traceback",
-        "types.TracebackType",
-        "TypeMap",
-        "AwaitableOrValue",
-        "DeferredFragmentRecord",
-        "DeferUsage",
-        "EnterLeaveVisitor",
-        "ExperimentalIncrementalExecutionResults",
-        "FieldGroup",
-        "FormattedIncrementalResult",
-        "FormattedPendingResult",
-        "FormattedSourceLocation",
-        "GraphQLAbstractType",
-        "GraphQLCompositeType",
-        "GraphQLEnumValueMap",
-        "GraphQLErrorExtensions",
-        "GraphQLFieldResolver",
-        "GraphQLInputType",
-        "GraphQLNullableType",
-        "GraphQLOutputType",
-        "GraphQLTypeResolver",
-        "GroupedFieldSet",
-        "IncrementalDataRecord",
-        "IncrementalResult",
-        "InitialResultRecord",
-        "Middleware",
-        "PendingResult",
-        "StreamItemsRecord",
-        "StreamRecord",
-        "SubsequentDataRecord",
-        "asyncio.events.AbstractEventLoop",
-        "collections.abc.MutableMapping",
-        "collections.abc.MutableSet",
-        "enum.Enum",
-        "graphql.execution.collect_fields.DeferUsage",
-        "graphql.execution.collect_fields.CollectFieldsResult",
-        "graphql.execution.collect_fields.FieldGroup",
-        "graphql.execution.execute.StreamArguments",
-        "graphql.execution.execute.StreamUsage",
-        "graphql.execution.map_async_iterable.map_async_iterable",
-        "graphql.execution.incremental_publisher.CompletedResult",
-        "graphql.execution.incremental_publisher.DeferredFragmentRecord",
-        "graphql.execution.incremental_publisher.DeferredGroupedFieldSetRecord",
-        "graphql.execution.incremental_publisher.FormattedCompletedResult",
-        "graphql.execution.incremental_publisher.FormattedPendingResult",
-        "graphql.execution.incremental_publisher.IncrementalPublisher",
-        "graphql.execution.incremental_publisher.InitialResultRecord",
-        "graphql.execution.incremental_publisher.PendingResult",
-        "graphql.execution.incremental_publisher.StreamItemsRecord",
-        "graphql.execution.incremental_publisher.StreamRecord",
-        "graphql.execution.Middleware",
-        "graphql.language.lexer.EscapeSequence",
-        "graphql.language.visitor.EnterLeaveVisitor",
-        "graphql.pyutils.ref_map.K",
-        "graphql.pyutils.ref_map.V",
-        "graphql.type.definition.GT_co",
-        "graphql.type.definition.GNT_co",
-        "graphql.type.definition.TContext",
-        "graphql.type.schema.InterfaceImplementations",
-        "graphql.validation.validation_context.VariableUsage",
-        "graphql.validation.rules.known_argument_names.KnownArgumentNamesOnDirectivesRule",
-        "graphql.validation.rules.provided_required_arguments.ProvidedRequiredArgumentsOnDirectivesRule",
-    ]
-)
+ignore_references = {
+    "GNT",
+    "GT",
+    "KT",
+    "T",
+    "VT",
+    "TContext",
+    "Enum",
+    "traceback",
+    "types.TracebackType",
+    "TypeMap",
+    "AwaitableOrValue",
+    "DeferredFragmentRecord",
+    "DeferUsage",
+    "EnterLeaveVisitor",
+    "ExperimentalIncrementalExecutionResults",
+    "FieldGroup",
+    "FormattedIncrementalResult",
+    "FormattedPendingResult",
+    "FormattedSourceLocation",
+    "GraphQLAbstractType",
+    "GraphQLCompositeType",
+    "GraphQLEnumValueMap",
+    "GraphQLErrorExtensions",
+    "GraphQLFieldResolver",
+    "GraphQLInputType",
+    "GraphQLNullableType",
+    "GraphQLOutputType",
+    "GraphQLTypeResolver",
+    "GroupedFieldSet",
+    "IncrementalDataRecord",
+    "IncrementalResult",
+    "InitialResultRecord",
+    "Middleware",
+    "PendingResult",
+    "StreamItemsRecord",
+    "StreamRecord",
+    "SubsequentDataRecord",
+    "asyncio.events.AbstractEventLoop",
+    "collections.abc.MutableMapping",
+    "collections.abc.MutableSet",
+    "enum.Enum",
+    "graphql.execution.build_field_plan.FieldGroup",
+    "graphql.execution.build_field_plan.FieldPlan",
+    "graphql.execution.collect_fields.DeferUsage",
+    "graphql.execution.execute.StreamArguments",
+    "graphql.execution.execute.StreamUsage",
+    "graphql.execution.map_async_iterable.map_async_iterable",
+    "graphql.execution.incremental_publisher.CompletedResult",
+    "graphql.execution.incremental_publisher.DeferredFragmentRecord",
+    "graphql.execution.incremental_publisher.DeferredGroupedFieldSetRecord",
+    "graphql.execution.incremental_publisher.FormattedCompletedResult",
+    "graphql.execution.incremental_publisher.FormattedPendingResult",
+    "graphql.execution.incremental_publisher.IncrementalPublisher",
+    "graphql.execution.incremental_publisher.InitialResultRecord",
+    "graphql.execution.incremental_publisher.PendingResult",
+    "graphql.execution.incremental_publisher.StreamItemsRecord",
+    "graphql.execution.incremental_publisher.StreamRecord",
+    "graphql.execution.Middleware",
+    "graphql.language.lexer.EscapeSequence",
+    "graphql.language.visitor.EnterLeaveVisitor",
+    "graphql.pyutils.ref_map.K",
+    "graphql.pyutils.ref_map.V",
+    "graphql.type.definition.GT_co",
+    "graphql.type.definition.GNT_co",
+    "graphql.type.definition.TContext",
+    "graphql.type.schema.InterfaceImplementations",
+    "graphql.validation.validation_context.VariableUsage",
+    "graphql.validation.rules.known_argument_names.KnownArgumentNamesOnDirectivesRule",
+    "graphql.validation.rules.provided_required_arguments.ProvidedRequiredArgumentsOnDirectivesRule",
+}
 
 ignore_references.update(__builtins__.keys())
 
@@ -228,10 +226,12 @@ def on_missing_reference(app, env, node, contnode):
     name = target.rsplit(".", 1)[-1]
     if name in ("GT", "GNT", "KT", "T", "VT"):
         return contnode
-    if typ == "obj":
-        if target.startswith("typing."):
-            if name in ("Any", "Optional", "Union"):
-                return contnode
+    if (
+        typ == "obj"
+        and target.startswith("typing.")
+        and name in ("Any", "Optional", "Union")
+    ):
+        return contnode
     if typ != "class":
         return None
     if "." in target:  # maybe too specific

src/graphql/execution/build_field_plan.py

@@ -0,0 +1,148 @@
+"""Build field plan"""
+
+from __future__ import annotations
+
+import sys
+from typing import TYPE_CHECKING, Dict, NamedTuple
+
+from ..pyutils import RefMap, RefSet
+from .collect_fields import DeferUsage, FieldDetails
+
+if TYPE_CHECKING:
+    from ..language import FieldNode
+
+try:
+    from typing import TypeAlias
+except ImportError:  # Python < 3.10
+    from typing_extensions import TypeAlias
+
+__all__ = [
+    "DeferUsageSet",
+    "FieldGroup",
+    "FieldPlan",
+    "GroupedFieldSet",
+    "NewGroupedFieldSetDetails",
+    "build_field_plan",
+]
+
+
+DeferUsageSet: TypeAlias = RefSet[DeferUsage]
+
+
+class FieldGroup(NamedTuple):
+    """A group of fields with defer usages."""
+
+    fields: list[FieldDetails]
+    defer_usages: DeferUsageSet | None = None
+    known_defer_usages: DeferUsageSet | None = None
+
+    def to_nodes(self) -> list[FieldNode]:
+        """Return the field nodes in this group."""
+        return [field_details.node for field_details in self.fields]
+
+
+if sys.version_info < (3, 9):
+    GroupedFieldSet: TypeAlias = Dict[str, FieldGroup]
+else:  # Python >= 3.9
+    GroupedFieldSet: TypeAlias = dict[str, FieldGroup]
+
+
+class NewGroupedFieldSetDetails(NamedTuple):
+    """Details of a new grouped field set."""
+
+    grouped_field_set: GroupedFieldSet
+    should_initiate_defer: bool
+
+
+class FieldPlan(NamedTuple):
+    """A plan for executing fields."""
+
+    grouped_field_set: GroupedFieldSet
+    new_grouped_field_set_details_map: RefMap[DeferUsageSet, NewGroupedFieldSetDetails]
+    new_defer_usages: list[DeferUsage]
+
+
+def build_field_plan(
+    fields: dict[str, list[FieldDetails]],
+    parent_defer_usages: DeferUsageSet | None = None,
+    known_defer_usages: DeferUsageSet | None = None,
+) -> FieldPlan:
+    """Build a plan for executing fields."""
+    if parent_defer_usages is None:
+        parent_defer_usages = RefSet()
+    if known_defer_usages is None:
+        known_defer_usages = RefSet()
+
+    new_defer_usages: RefSet[DeferUsage] = RefSet()
+    new_known_defer_usages: RefSet[DeferUsage] = RefSet(known_defer_usages)
+
+    grouped_field_set: GroupedFieldSet = {}
+
+    new_grouped_field_set_details_map: RefMap[
+        DeferUsageSet, NewGroupedFieldSetDetails
+    ] = RefMap()
+
+    map_: dict[str, tuple[DeferUsageSet, list[FieldDetails]]] = {}
+
+    for response_key, field_details_list in fields.items():
+        defer_usage_set: RefSet[DeferUsage] = RefSet()
+        in_original_result = False
+        for field_details in field_details_list:
+            defer_usage = field_details.defer_usage
+            if defer_usage is None:
+                in_original_result = True
+                continue
+            defer_usage_set.add(defer_usage)
+            if defer_usage not in known_defer_usages:
+                new_defer_usages.add(defer_usage)
+                new_known_defer_usages.add(defer_usage)
+        if in_original_result:
+            defer_usage_set.clear()
+        else:
+            defer_usage_set -= {
+                defer_usage
+                for defer_usage in defer_usage_set
+                if any(
+                    ancestor in defer_usage_set for ancestor in defer_usage.ancestors
+                )
+            }
+        map_[response_key] = (defer_usage_set, field_details_list)
+
+    for response_key, [defer_usage_set, field_details_list] in map_.items():
+        if defer_usage_set == parent_defer_usages:
+            field_group = grouped_field_set.get(response_key)
+            if field_group is None:  # pragma: no cover else
+                field_group = FieldGroup([], defer_usage_set, new_known_defer_usages)
+                grouped_field_set[response_key] = field_group
+            field_group.fields.extend(field_details_list)
+            continue
+
+        for (
+            new_grouped_field_set_defer_usage_set,
+            new_grouped_field_set_details,
+        ) in new_grouped_field_set_details_map.items():
+            if new_grouped_field_set_defer_usage_set == defer_usage_set:
+                new_grouped_field_set = new_grouped_field_set_details.grouped_field_set
+                break
+        else:
+            new_grouped_field_set = {}
+            new_grouped_field_set_details = NewGroupedFieldSetDetails(
+                new_grouped_field_set,
+                any(
+                    defer_usage not in parent_defer_usages
+                    for defer_usage in defer_usage_set
+                ),
+            )
+            new_grouped_field_set_details_map[defer_usage_set] = (
+                new_grouped_field_set_details
+            )
+
+        field_group = new_grouped_field_set.get(response_key)
+        if field_group is None:  # pragma: no cover else
+            field_group = FieldGroup([], defer_usage_set, new_known_defer_usages)
+            new_grouped_field_set[response_key] = field_group
+        field_group.fields.extend(field_details_list)
+
+    return FieldPlan(
+        grouped_field_set, new_grouped_field_set_details_map, list(new_defer_usages)
+    )

src/graphql/execution/collect_fields.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-import sys
-from typing import Any, Dict, NamedTuple, Union, cast
+from collections import defaultdict
+from typing import Any, NamedTuple
 
 from ..language import (
     FieldNode,
@@ -14,7 +14,6 @@
     OperationType,
     SelectionSetNode,
 )
-from ..pyutils import RefMap, RefSet
 from ..type import (
     GraphQLDeferDirective,
     GraphQLIncludeDirective,
@@ -26,81 +25,37 @@
 from ..utilities.type_from_ast import type_from_ast
 from .values import get_directive_values
 
-try:
-    from typing import TypeAlias
-except ImportError:  # Python < 3.10
-    from typing_extensions import TypeAlias
-
-
 __all__ = [
-    "NON_DEFERRED_TARGET_SET",
     "CollectFieldsContext",
-    "CollectFieldsResult",
     "DeferUsage",
-    "DeferUsageSet",
     "FieldDetails",
-    "FieldGroup",
-    "GroupedFieldSetDetails",
-    "Target",
-    "TargetSet",
     "collect_fields",
     "collect_subfields",
 ]
 
 
 class DeferUsage(NamedTuple):
-    """An optionally labelled list of ancestor targets."""
+    """An optionally labelled linked list of defer usages."""
 
     label: str | None
-    ancestors: list[Target]
-
-
-Target: TypeAlias = Union[DeferUsage, None]
+    parent_defer_usage: DeferUsage | None
 
-TargetSet: TypeAlias = RefSet[Target]
-DeferUsageSet: TypeAlias = RefSet[DeferUsage]
-
-
-NON_DEFERRED_TARGET_SET: TargetSet = RefSet([None])
+    @property
+    def ancestors(self) -> list[DeferUsage]:
+        """Get the ancestors of this defer usage."""
+        ancestors: list[DeferUsage] = []
+        parent_defer_usage = self.parent_defer_usage
+        while parent_defer_usage is not None:
+            ancestors.append(parent_defer_usage)
+            parent_defer_usage = parent_defer_usage.parent_defer_usage
+        return ancestors[::-1]
 
 
 class FieldDetails(NamedTuple):
-    """A field node and its target."""
+    """A field node and its defer usage."""
 
     node: FieldNode
-    target: Target
-
-
-class FieldGroup(NamedTuple):
-    """A group of fields that share the same target set."""
-
-    fields: list[FieldDetails]
-    targets: TargetSet
-
-    def to_nodes(self) -> list[FieldNode]:
-        """Return the field nodes in this group."""
-        return [field_details.node for field_details in self.fields]
-
-
-if sys.version_info < (3, 9):
-    GroupedFieldSet: TypeAlias = Dict[str, FieldGroup]
-else:  # Python >= 3.9
-    GroupedFieldSet: TypeAlias = dict[str, FieldGroup]
-
-
-class GroupedFieldSetDetails(NamedTuple):
-    """A grouped field set with defer info."""
-
-    grouped_field_set: GroupedFieldSet
-    should_initiate_defer: bool
-
-
-class CollectFieldsResult(NamedTuple):
-    """Collected fields and deferred usages."""
-
-    grouped_field_set: GroupedFieldSet
-    new_grouped_field_set_details: RefMap[DeferUsageSet, GroupedFieldSetDetails]
-    new_defer_usages: list[DeferUsage]
+    defer_usage: DeferUsage | None
 
 
 class CollectFieldsContext(NamedTuple):
@@ -111,9 +66,6 @@ class CollectFieldsContext(NamedTuple):
     variable_values: dict[str, Any]
     operation: OperationDefinitionNode
     runtime_type: GraphQLObjectType
-    targets_by_key: dict[str, TargetSet]
-    fields_by_target: RefMap[Target, dict[str, list[FieldNode]]]
-    new_defer_usages: list[DeferUsage]
     visited_fragment_names: set[str]
 
 
@@ -123,7 +75,7 @@ def collect_fields(
     variable_values: dict[str, Any],
     runtime_type: GraphQLObjectType,
     operation: OperationDefinitionNode,
-) -> CollectFieldsResult:
+) -> dict[str, list[FieldDetails]]:
     """Collect fields.
 
     Given a selection_set, collects all the fields and returns them.
@@ -134,23 +86,18 @@ def collect_fields(
 
     For internal use only.
     """
+    grouped_field_set: dict[str, list[FieldDetails]] = defaultdict(list)
     context = CollectFieldsContext(
         schema,
         fragments,
         variable_values,
         operation,
         runtime_type,
-        {},
-        RefMap(),
-        [],
         set(),
     )
-    collect_fields_impl(context, operation.selection_set)
 
-    return CollectFieldsResult(
-        *build_grouped_field_sets(context.targets_by_key, context.fields_by_target),
-        context.new_defer_usages,
-    )
+    collect_fields_impl(context, operation.selection_set, grouped_field_set)
+    return grouped_field_set
 
 
 def collect_subfields(
@@ -159,8 +106,8 @@ def collect_subfields(
     variable_values: dict[str, Any],
     operation: OperationDefinitionNode,
     return_type: GraphQLObjectType,
-    field_group: FieldGroup,
-) -> CollectFieldsResult:
+    field_details: list[FieldDetails],
+) -> dict[str, list[FieldDetails]]:
     """Collect subfields.
 
     Given a list of field nodes, collects all the subfields of the passed in fields,
@@ -178,30 +125,29 @@ def collect_subfields(
         variable_values,
         operation,
         return_type,
-        {},
-        RefMap(),
-        [],
         set(),
     )
+    sub_grouped_field_set: dict[str, list[FieldDetails]] = defaultdict(list)
 
-    for field_details in field_group.fields:
-        node = field_details.node
+    for field_detail in field_details:
+        node = field_detail.node
         if node.selection_set:
-            collect_fields_impl(context, node.selection_set, field_details.target)
+            collect_fields_impl(
+                context,
+                node.selection_set,
+                sub_grouped_field_set,
+                field_detail.defer_usage,
+            )
 
-    return CollectFieldsResult(
-        *build_grouped_field_sets(
-            context.targets_by_key, context.fields_by_target, field_group.targets
-        ),
-        context.new_defer_usages,
-    )
+    return sub_grouped_field_set
 
 
 def collect_fields_impl(
     context: CollectFieldsContext,
     selection_set: SelectionSetNode,
-    parent_target: Target | None = None,
-    new_target: Target | None = None,
+    grouped_field_set: dict[str, list[FieldDetails]],
+    parent_defer_usage: DeferUsage | None = None,
+    defer_usage: DeferUsage | None = None,
 ) -> None:
     """Collect fields (internal implementation)."""
     (
@@ -210,97 +156,71 @@ def collect_fields_impl(
         variable_values,
         operation,
         runtime_type,
-        targets_by_key,
-        fields_by_target,
-        new_defer_usages,
         visited_fragment_names,
     ) = context
 
-    ancestors: list[Target]
-
     for selection in selection_set.selections:
         if isinstance(selection, FieldNode):
             if not should_include_node(variable_values, selection):
                 continue
             key = get_field_entry_key(selection)
-            target = new_target or parent_target
-            key_targets = targets_by_key.get(key)
-            if key_targets is None:
-                key_targets = RefSet([target])
-                targets_by_key[key] = key_targets
-            else:
-                key_targets.add(target)
-            target_fields = fields_by_target.get(target)
-            if target_fields is None:
-                fields_by_target[target] = {key: [selection]}
-            else:
-                field_nodes = target_fields.get(key)
-                if field_nodes is None:
-                    target_fields[key] = [selection]
-                else:
-                    field_nodes.append(selection)
+            grouped_field_set[key].append(
+                FieldDetails(selection, defer_usage or parent_defer_usage)
+            )
         elif isinstance(selection, InlineFragmentNode):
             if not should_include_node(
                 variable_values, selection
             ) or not does_fragment_condition_match(schema, selection, runtime_type):
                 continue
 
-            defer = get_defer_values(operation, variable_values, selection)
-
-            if defer:
-                ancestors = (
-                    [None]
-                    if parent_target is None
-                    else [parent_target, *parent_target.ancestors]
-                )
-                target = DeferUsage(defer.label, ancestors)
-                new_defer_usages.append(target)
-            else:
-                target = new_target
+            new_defer_usage = get_defer_usage(
+                operation, variable_values, selection, parent_defer_usage
+            )
 
-            collect_fields_impl(context, selection.selection_set, parent_target, target)
+            collect_fields_impl(
+                context,
+                selection.selection_set,
+                grouped_field_set,
+                parent_defer_usage,
+                new_defer_usage or defer_usage,
+            )
         elif isinstance(selection, FragmentSpreadNode):  # pragma: no cover else
             frag_name = selection.name.value
 
-            if not should_include_node(variable_values, selection):
-                continue
+            new_defer_usage = get_defer_usage(
+                operation, variable_values, selection, parent_defer_usage
+            )
 
-            defer = get_defer_values(operation, variable_values, selection)
-            if frag_name in visited_fragment_names and not defer:
+            if new_defer_usage is None and (
+                frag_name in visited_fragment_names
+                or not should_include_node(variable_values, selection)
+            ):
                 continue
 
             fragment = fragments.get(frag_name)
-            if not fragment or not does_fragment_condition_match(
+            if fragment is None or not does_fragment_condition_match(
                 schema, fragment, runtime_type
             ):
                 continue
 
-            if defer:
-                ancestors = (
-                    [None]
-                    if parent_target is None
-                    else [parent_target, *parent_target.ancestors]
-                )
-                target = DeferUsage(defer.label, ancestors)
-                new_defer_usages.append(target)
-            else:
+            if new_defer_usage is None:
                 visited_fragment_names.add(frag_name)
-                target = new_target
 
-            collect_fields_impl(context, fragment.selection_set, parent_target, target)
-
-
-class DeferValues(NamedTuple):
-    """Values of an active defer directive."""
-
-    label: str | None
+            collect_fields_impl(
+                context,
+                fragment.selection_set,
+                grouped_field_set,
+                parent_defer_usage,
+                new_defer_usage or defer_usage,
+            )
 
 
-def get_defer_values(
+def get_defer_usage(
     operation: OperationDefinitionNode,
     variable_values: dict[str, Any],
     node: FragmentSpreadNode | InlineFragmentNode,
-) -> DeferValues | None:
+    parent_defer_usage: DeferUsage | None,
+) -> DeferUsage | None:
     """Get values of defer directive if active.
 
     Returns an object containing the `@defer` arguments if a field should be
@@ -319,7 +239,7 @@ def get_defer_values(
         )
         raise TypeError(msg)
 
-    return DeferValues(defer.get("label"))
+    return DeferUsage(defer.get("label"), parent_defer_usage)
 
 
 def should_include_node(
@@ -360,111 +280,3 @@ def does_fragment_condition_match(
 def get_field_entry_key(node: FieldNode) -> str:
     """Implement the logic to compute the key of a given field's entry"""
     return node.alias.value if node.alias else node.name.value
-
-
-def build_grouped_field_sets(
-    targets_by_key: dict[str, TargetSet],
-    fields_by_target: RefMap[Target, dict[str, list[FieldNode]]],
-    parent_targets: TargetSet = NON_DEFERRED_TARGET_SET,
-) -> tuple[GroupedFieldSet, RefMap[DeferUsageSet, GroupedFieldSetDetails]]:
-    """Build grouped field sets."""
-    parent_target_keys, target_set_details_map = get_target_set_details(
-        targets_by_key, parent_targets
-    )
-
-    grouped_field_set = (
-        get_ordered_grouped_field_set(
-            parent_target_keys, parent_targets, targets_by_key, fields_by_target
-        )
-        if parent_target_keys
-        else {}
-    )
-
-    new_grouped_field_set_details: RefMap[DeferUsageSet, GroupedFieldSetDetails] = (
-        RefMap()
-    )
-
-    for masking_targets, target_set_details in target_set_details_map.items():
-        keys, should_initiate_defer = target_set_details
-
-        new_grouped_field_set = get_ordered_grouped_field_set(
-            keys, masking_targets, targets_by_key, fields_by_target
-        )
-
-        # All TargetSets that causes new grouped field sets consist only of DeferUsages
-        # and have should_initiate_defer defined
-
-        new_grouped_field_set_details[cast("DeferUsageSet", masking_targets)] = (
-            GroupedFieldSetDetails(new_grouped_field_set, should_initiate_defer)
-        )
-
-    return grouped_field_set, new_grouped_field_set_details
-
-
-class TargetSetDetails(NamedTuple):
-    """A set of target keys with defer info."""
-
-    keys: set[str]
-    should_initiate_defer: bool
-
-
-def get_target_set_details(
-    targets_by_key: dict[str, TargetSet], parent_targets: TargetSet
-) -> tuple[set[str], RefMap[TargetSet, TargetSetDetails]]:
-    """Get target set details."""
-    parent_target_keys: set[str] = set()
-    target_set_details_map: RefMap[TargetSet, TargetSetDetails] = RefMap()
-
-    for response_key, targets in targets_by_key.items():
-        masking_target_list: list[Target] = []
-        for target in targets:
-            if not target or all(
-                ancestor not in targets for ancestor in target.ancestors
-            ):
-                masking_target_list.append(target)
-
-        masking_targets: TargetSet = RefSet(masking_target_list)
-        if masking_targets == parent_targets:
-            parent_target_keys.add(response_key)
-            continue
-
-        for target_set, target_set_details in target_set_details_map.items():
-            if target_set == masking_targets:
-                target_set_details.keys.add(response_key)
-                break
-        else:
-            target_set_details = TargetSetDetails(
-                {response_key},
-                any(
-                    defer_usage not in parent_targets for defer_usage in masking_targets
-                ),
-            )
-            target_set_details_map[masking_targets] = target_set_details
-
-    return parent_target_keys, target_set_details_map
-
-
-def get_ordered_grouped_field_set(
-    keys: set[str],
-    masking_targets: TargetSet,
-    targets_by_key: dict[str, TargetSet],
-    fields_by_target: RefMap[Target, dict[str, list[FieldNode]]],
-) -> GroupedFieldSet:
-    """Get ordered grouped field set."""
-    grouped_field_set: GroupedFieldSet = {}
-
-    first_target = next(iter(masking_targets))
-    first_fields = fields_by_target[first_target]
-    for key in list(first_fields):
-        if key in keys:
-            field_group = grouped_field_set.get(key)
-            if field_group is None:  # pragma: no cover else
-                field_group = FieldGroup([], masking_targets)
-                grouped_field_set[key] = field_group
-            for target in targets_by_key[key]:
-                fields_for_target = fields_by_target[target]
-                nodes = fields_for_target[key]
-                del fields_for_target[key]
-                field_group.fields.extend(FieldDetails(node, target) for node in nodes)
-
-    return grouped_field_set

src/graphql/execution/execute.py

@@ -40,6 +40,7 @@
 from ..error import GraphQLError, located_error
 from ..language import (
     DocumentNode,
+    FieldNode,
     FragmentDefinitionNode,
     OperationDefinitionNode,
     OperationType,
@@ -75,18 +76,15 @@
     is_object_type,
 )
 from .async_iterables import map_async_iterable
-from .collect_fields import (
-    NON_DEFERRED_TARGET_SET,
-    CollectFieldsResult,
-    DeferUsage,
+from .build_field_plan import (
     DeferUsageSet,
-    FieldDetails,
     FieldGroup,
+    FieldPlan,
     GroupedFieldSet,
-    GroupedFieldSetDetails,
-    collect_fields,
-    collect_subfields,
+    NewGroupedFieldSetDetails,
+    build_field_plan,
 )
+from .collect_fields import DeferUsage, FieldDetails, collect_fields, collect_subfields
 from .incremental_publisher import (
     ASYNC_DELAY,
     DeferredFragmentRecord,
@@ -208,7 +206,7 @@ def __init__(
         if is_awaitable:
             self.is_awaitable = is_awaitable
         self._canceled_iterators: set[AsyncIterator] = set()
-        self._subfields_cache: dict[tuple, CollectFieldsResult] = {}
+        self._field_plan_cache: dict[tuple, FieldPlan] = {}
         self._tasks: set[Awaitable] = set()
         self._stream_usages: RefMap[FieldGroup, StreamUsage] = RefMap()
 
@@ -326,10 +324,11 @@ def execute_operation(
             )
             raise GraphQLError(msg, operation)
 
-        grouped_field_set, new_grouped_field_set_details, new_defer_usages = (
-            collect_fields(
-                schema, self.fragments, self.variable_values, root_type, operation
-            )
+        fields = collect_fields(
+            schema, self.fragments, self.variable_values, root_type, operation
+        )
+        grouped_field_set, new_grouped_field_set_details_map, new_defer_usages = (
+            build_field_plan(fields)
         )
 
         incremental_publisher = self.incremental_publisher
@@ -341,7 +340,7 @@ def execute_operation(
 
         new_deferred_grouped_field_set_records = add_new_deferred_grouped_field_sets(
             incremental_publisher,
-            new_grouped_field_set_details,
+            new_grouped_field_set_details_map,
             new_defer_map,
             path,
         )
@@ -502,7 +501,9 @@ def execute_field(
         if self.middleware_manager:
             resolve_fn = self.middleware_manager.get_field_resolver(resolve_fn)
 
-        info = self.build_resolve_info(field_def, field_group, parent_type, path)
+        info = self.build_resolve_info(
+            field_def, field_group.to_nodes(), parent_type, path
+        )
 
         # Get the resolve function, regardless of if its result is normal or abrupt
         # (error).
@@ -575,7 +576,7 @@ async def await_completed() -> Any:
     def build_resolve_info(
         self,
         field_def: GraphQLField,
-        field_group: FieldGroup,
+        field_nodes: list[FieldNode],
         parent_type: GraphQLObjectType,
         path: Path,
     ) -> GraphQLResolveInfo:
@@ -586,8 +587,8 @@ def build_resolve_info(
         # The resolve function's first argument is a collection of information about
         # the current execution state.
         return GraphQLResolveInfo(
-            field_group.fields[0].node.name.value,
-            field_group.to_nodes(),
+            field_nodes[0].name.value,
+            field_nodes,
             field_def.type,
             parent_type,
             path,
@@ -807,8 +808,7 @@ def get_stream_usage(
             [
                 FieldDetails(field_details.node, None)
                 for field_details in field_group.fields
-            ],
-            NON_DEFERRED_TARGET_SET,
+            ]
         )
 
         stream_usage = StreamUsage(
@@ -1273,8 +1273,8 @@ def collect_and_execute_subfields(
         defer_map: RefMap[DeferUsage, DeferredFragmentRecord],
     ) -> AwaitableOrValue[dict[str, Any]]:
         """Collect sub-fields to execute to complete this value."""
-        grouped_field_set, new_grouped_field_set_details, new_defer_usages = (
-            self.collect_subfields(return_type, field_group)
+        grouped_field_set, new_grouped_field_set_details_map, new_defer_usages = (
+            self.build_sub_field_plan(return_type, field_group)
         )
 
         incremental_publisher = self.incremental_publisher
@@ -1286,7 +1286,10 @@ def collect_and_execute_subfields(
             path,
         )
         new_deferred_grouped_field_set_records = add_new_deferred_grouped_field_sets(
-            incremental_publisher, new_grouped_field_set_details, new_defer_map, path
+            incremental_publisher,
+            new_grouped_field_set_details_map,
+            new_defer_map,
+            path,
         )
 
         sub_fields = self.execute_fields(
@@ -1308,17 +1311,16 @@ def collect_and_execute_subfields(
 
         return sub_fields
 
-    def collect_subfields(
+    def build_sub_field_plan(
         self, return_type: GraphQLObjectType, field_group: FieldGroup
-    ) -> CollectFieldsResult:
+    ) -> FieldPlan:
         """Collect subfields.
 
-        A cached collection of relevant subfields with regard to the return type is
-        kept in the execution context as ``_subfields_cache``. This ensures the
-        subfields are not repeatedly calculated, which saves overhead when resolving
-        lists of values.
+        A memoized function for building subfield plans with regard to the return type.
+        Memoizing ensures the  subfields are not repeatedly calculated, which saves
+        overhead when resolving lists of values.
         """
-        cache = self._subfields_cache
+        cache = self._field_plan_cache
         # We cannot use the field_group itself as key for the cache, since it
         # is not hashable as a list. We also do not want to use the field_group
         # itself (converted to a tuple) as keys, since hashing them is slow.
@@ -1331,18 +1333,21 @@ def collect_subfields(
             if len(field_group) == 1  # optimize most frequent case
             else (return_type, *map(id, field_group))
         )
-        sub_fields_and_patches = cache.get(key)
-        if sub_fields_and_patches is None:
-            sub_fields_and_patches = collect_subfields(
+        plan = cache.get(key)
+        if plan is None:
+            sub_fields = collect_subfields(
                 self.schema,
                 self.fragments,
                 self.variable_values,
                 self.operation,
                 return_type,
-                field_group,
+                field_group.fields,
+            )
+            plan = build_field_plan(
+                sub_fields, field_group.defer_usages, field_group.known_defer_usages
             )
-            cache[key] = sub_fields_and_patches
-        return sub_fields_and_patches
+            cache[key] = plan
+        return plan
 
     def map_source_to_response(
         self, result_or_stream: ExecutionResult | AsyncIterable[Any]
@@ -1954,14 +1959,10 @@ def add_new_deferred_fragments(
     new_defer_map = RefMap() if defer_map is None else RefMap(defer_map.items())
 
     # For each new DeferUsage object:
-    for defer_usage in new_defer_usages:
-        ancestors = defer_usage.ancestors
-        parent_defer_usage = ancestors[0] if ancestors else None
-
-        # If the parent target is defined, the parent target is a DeferUsage object
-        # and the parent result record is the DeferredFragmentRecord corresponding
-        # to that DeferUsage.
-        # If the parent target is not defined, the parent result record is either:
+    for new_defer_usage in new_defer_usages:
+        parent_defer_usage = new_defer_usage.parent_defer_usage
+
+        # If the parent defer usage is not defined, the parent result record is either:
         #  - the InitialResultRecord, or
         #  - a StreamItemsRecord, as `@defer` may be nested under `@stream`.
         parent = (
@@ -1975,15 +1976,15 @@ def add_new_deferred_fragments(
         )
 
         # Instantiate the new record.
-        deferred_fragment_record = DeferredFragmentRecord(path, defer_usage.label)
+        deferred_fragment_record = DeferredFragmentRecord(path, new_defer_usage.label)
 
         # Report the new record to the Incremental Publisher.
         incremental_publisher.report_new_defer_fragment_record(
             deferred_fragment_record, parent
         )
 
         # Update the map.
-        new_defer_map[defer_usage] = deferred_fragment_record
+        new_defer_map[new_defer_usage] = deferred_fragment_record
 
     return new_defer_map
 
@@ -1997,24 +1998,26 @@ def deferred_fragment_record_from_defer_usage(
 
 def add_new_deferred_grouped_field_sets(
     incremental_publisher: IncrementalPublisher,
-    new_grouped_field_set_details: Mapping[DeferUsageSet, GroupedFieldSetDetails],
+    new_grouped_field_set_details_map: Mapping[
+        DeferUsageSet, NewGroupedFieldSetDetails
+    ],
     defer_map: RefMap[DeferUsage, DeferredFragmentRecord],
     path: Path | None = None,
 ) -> list[DeferredGroupedFieldSetRecord]:
     """Add new deferred grouped field sets to the defer map."""
     new_deferred_grouped_field_set_records: list[DeferredGroupedFieldSetRecord] = []
 
     for (
-        new_grouped_field_set_defer_usages,
-        grouped_field_set_details,
-    ) in new_grouped_field_set_details.items():
+        defer_usage_set,
+        [grouped_field_set, should_initiate_defer],
+    ) in new_grouped_field_set_details_map.items():
         deferred_fragment_records = get_deferred_fragment_records(
-            new_grouped_field_set_defer_usages, defer_map
+            defer_usage_set, defer_map
         )
         deferred_grouped_field_set_record = DeferredGroupedFieldSetRecord(
             deferred_fragment_records,
-            grouped_field_set_details.grouped_field_set,
-            grouped_field_set_details.should_initiate_defer,
+            grouped_field_set,
+            should_initiate_defer,
             path,
         )
         incremental_publisher.report_new_deferred_grouped_filed_set_record(
@@ -2292,34 +2295,34 @@ def execute_subscription(
         msg = "Schema is not configured to execute subscription operation."
         raise GraphQLError(msg, context.operation)
 
-    grouped_field_set = collect_fields(
+    fields = collect_fields(
         schema,
         context.fragments,
         context.variable_values,
         root_type,
         context.operation,
-    ).grouped_field_set
-    first_root_field = next(iter(grouped_field_set.items()))
-    response_name, field_group = first_root_field
-    field_name = field_group.fields[0].node.name.value
+    )
+
+    first_root_field = next(iter(fields.items()))
+    response_name, field_details_list = first_root_field
+    field_name = field_details_list[0].node.name.value
     field_def = schema.get_field(root_type, field_name)
 
+    field_nodes = [field_details.node for field_details in field_details_list]
     if not field_def:
         msg = f"The subscription field '{field_name}' is not defined."
-        raise GraphQLError(msg, field_group.to_nodes())
+        raise GraphQLError(msg, field_nodes)
 
     path = Path(None, response_name, root_type.name)
-    info = context.build_resolve_info(field_def, field_group, root_type, path)
+    info = context.build_resolve_info(field_def, field_nodes, root_type, path)
 
     # Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
     # It differs from "ResolveFieldValue" due to providing a different `resolveFn`.
 
     try:
         # Build a dictionary of arguments from the field.arguments AST, using the
         # variables scope to fulfill any variable references.
-        args = get_argument_values(
-            field_def, field_group.fields[0].node, context.variable_values
-        )
+        args = get_argument_values(field_def, field_nodes[0], context.variable_values)
 
         # Call the `subscribe()` resolver or the default resolver to produce an
         # AsyncIterable yielding raw payloads.
@@ -2332,16 +2335,14 @@ async def await_result() -> AsyncIterable[Any]:
                 try:
                     return assert_event_stream(await result)
                 except Exception as error:
-                    raise located_error(
-                        error, field_group.to_nodes(), path.as_list()
-                    ) from error
+                    raise located_error(error, field_nodes, path.as_list()) from error
 
             return await_result()
 
         return assert_event_stream(result)
 
     except Exception as error:
-        raise located_error(error, field_group.to_nodes(), path.as_list()) from error
+        raise located_error(error, field_nodes, path.as_list()) from error
 
 
 def assert_event_stream(result: Any) -> AsyncIterable:

src/graphql/execution/incremental_publisher.py

@@ -26,7 +26,7 @@
 if TYPE_CHECKING:
     from ..error import GraphQLError, GraphQLFormattedError
     from ..pyutils import Path
-    from .collect_fields import GroupedFieldSet
+    from .build_field_plan import GroupedFieldSet
 
 __all__ = [
     "ASYNC_DELAY",

src/graphql/validation/rules/single_field_subscriptions.py

@@ -5,7 +5,7 @@
 from typing import Any
 
 from ...error import GraphQLError
-from ...execution.collect_fields import FieldGroup, collect_fields
+from ...execution.collect_fields import FieldDetails, collect_fields
 from ...language import (
     FieldNode,
     FragmentDefinitionNode,
@@ -17,8 +17,8 @@
 __all__ = ["SingleFieldSubscriptionsRule"]
 
 
-def to_nodes(field_group: FieldGroup) -> list[FieldNode]:
-    return [field_details.node for field_details in field_group.fields]
+def to_nodes(field_details_list: list[FieldDetails]) -> list[FieldNode]:
+    return [field_details.node for field_details in field_details_list]
 
 
 class SingleFieldSubscriptionsRule(ValidationRule):
@@ -46,15 +46,15 @@ def enter_operation_definition(
                 for definition in document.definitions
                 if isinstance(definition, FragmentDefinitionNode)
             }
-            grouped_field_set = collect_fields(
+            fields = collect_fields(
                 schema,
                 fragments,
                 variable_values,
                 subscription_type,
                 node,
-            ).grouped_field_set
-            if len(grouped_field_set) > 1:
-                field_groups = list(grouped_field_set.values())
+            )
+            if len(fields) > 1:
+                field_groups = list(fields.values())
                 extra_field_groups = field_groups[1:]
                 extra_field_selection = [
                     node
@@ -72,7 +72,7 @@ def enter_operation_definition(
                         extra_field_selection,
                     )
                 )
-            for field_group in grouped_field_set.values():
+            for field_group in fields.values():
                 field_name = to_nodes(field_group)[0].name.value
                 if field_name.startswith("__"):
                     self.report_error(