deduplicate overlapping deferred and non-deferred fragment spreads

Only the fragment-spread deduplication from the upstream commit was missing:
visited_fragment_names becomes a dict[str, bool] tracking whether a fragment
was visited as deferred, so a deferred spread is skipped once visited at all
and a non-deferred spread only when previously visited as non-deferred.

Replicates graphql/graphql-js@8e01e23

Author: Cito

src/graphql/execution/collect_fields.py

@@ -82,7 +82,7 @@ class CollectFieldsContext(NamedTuple):
     variable_values: VariableValues
     operation: OperationDefinitionNode
     runtime_type: GraphQLObjectType
-    visited_fragment_names: set[str]
+    visited_fragment_names: dict[str, bool]
     hide_suggestions: bool
     forbidden_directive_instances: list[DirectiveNode]
     forbid_skip_and_include: bool
@@ -123,7 +123,7 @@ def collect_fields(
         variable_values,
         operation,
         runtime_type,
-        set(),
+        {},
         hide_suggestions,
         [],
         forbid_skip_and_include,
@@ -163,7 +163,7 @@ def collect_subfields(
         variable_values,
         operation,
         return_type,
-        set(),
+        {},
         hide_suggestions,
         [],
         False,
@@ -255,6 +255,17 @@ def collect_fields_impl(
         elif isinstance(selection, FragmentSpreadNode):  # pragma: no branch
             frag_name = selection.name.value
 
+            if not should_include_node(
+                context, selection, variable_values, fragment_variable_values
+            ):
+                continue
+
+            fragment = fragments.get(frag_name)
+            if fragment is None or not does_fragment_condition_match(
+                schema, fragment.definition, runtime_type
+            ):
+                continue
+
             new_defer_usage = get_defer_usage(
                 operation,
                 variable_values,
@@ -263,19 +274,24 @@ def collect_fields_impl(
                 defer_usage,
             )
 
-            if new_defer_usage is None and (
-                frag_name in visited_fragment_names
-                or not should_include_node(
-                    context, selection, variable_values, fragment_variable_values
-                )
-            ):
-                continue
+            visited_as_deferred = visited_fragment_names.get(frag_name)
 
-            fragment = fragments.get(frag_name)
-            if fragment is None or not does_fragment_condition_match(
-                schema, fragment.definition, runtime_type
-            ):
-                continue
+            if new_defer_usage is None:
+                # If this spread is not deferred, it may be skipped when already
+                # visited as a non-deferred spread. If it was previously visited as
+                # a deferred spread, it must be revisited.
+                if visited_as_deferred is False:
+                    continue
+                visited_fragment_names[frag_name] = False
+                maybe_new_defer_usage = defer_usage
+            else:
+                # If this spread is deferred, it can be skipped if it has already
+                # been visited.
+                if visited_as_deferred is not None:
+                    continue
+                visited_fragment_names[frag_name] = True
+                new_defer_usages.append(new_defer_usage)
+                maybe_new_defer_usage = new_defer_usage
 
             fragment_variable_signatures = fragment.variable_signatures
             new_fragment_variable_values: FragmentVariableValues | None = None
@@ -288,26 +304,14 @@ def collect_fields_impl(
                     hide_suggestions,
                 )
 
-            if new_defer_usage is None:
-                visited_fragment_names.add(frag_name)
-                collect_fields_impl(
-                    context,
-                    fragment.definition.selection_set,
-                    grouped_field_set,
-                    new_defer_usages,
-                    defer_usage,
-                    new_fragment_variable_values,
-                )
-            else:
-                new_defer_usages.append(new_defer_usage)
-                collect_fields_impl(
-                    context,
-                    fragment.definition.selection_set,
-                    grouped_field_set,
-                    new_defer_usages,
-                    new_defer_usage,
-                    new_fragment_variable_values,
-                )
+            collect_fields_impl(
+                context,
+                fragment.definition.selection_set,
+                grouped_field_set,
+                new_defer_usages,
+                maybe_new_defer_usage,
+                new_fragment_variable_values,
+            )
 
 
 def get_defer_usage(

tests/execution/test_collect_fields.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from graphql.execution import Executor
+from graphql.execution.collect_fields import collect_fields
+from graphql.language import parse
+from graphql.utilities import build_schema
+
+schema = build_schema(
+    """
+    type Query {
+      field: String
+    }
+    """
+)
+
+
+def collect_root_fields(query: str):
+    executor = Executor.build(schema, parse(query))
+
+    assert isinstance(executor, Executor)
+
+    query_type = schema.query_type
+    assert query_type is not None
+
+    return collect_fields(
+        executor.schema,
+        executor.fragments,
+        executor.variable_values,
+        query_type,
+        executor.operation,
+    )
+
+
+def describe_collect_fields():
+    def describe_overlapping_fragment_spreads():
+        def not_collect_deferred_spread_after_non_deferred_collected():
+            """Should not collect a deferred spread after a non-deferred."""
+            collected = collect_root_fields(
+                """
+                query {
+                  ...FragmentName
+                  ...FragmentName @defer
+                }
+                fragment FragmentName on Query {
+                  field
+                }
+                """
+            )
+
+            assert len(collected.new_defer_usages) == 0
+
+        def not_collect_deferred_spread_after_deferred_collected():
+            """Should not collect a deferred spread after a deferred one."""
+            collected = collect_root_fields(
+                """
+                query {
+                  ...FragmentName @defer
+                  ...FragmentName @defer
+                }
+                fragment FragmentName on Query {
+                  field
+                }
+                """
+            )
+
+            assert len(collected.new_defer_usages) == 1
+
+        def collect_non_deferred_spread_after_deferred_collected():
+            """Should collect a non-deferred spread after a deferred one."""
+            collected = collect_root_fields(
+                """
+                query {
+                  ...FragmentName @defer
+                  ...FragmentName
+                }
+                fragment FragmentName on Query {
+                  field
+                }
+                """
+            )
+
+            field_details_list = collected.grouped_field_set.get("field")
+
+            assert field_details_list is not None
+            assert len(field_details_list) == 2
+
+        def not_collect_non_deferred_spread_after_non_deferred_collected():
+            """Should not collect a non-deferred spread after a non-deferred."""
+            collected = collect_root_fields(
+                """
+                query {
+                  ...FragmentName
+                  ...FragmentName
+                }
+                fragment FragmentName on Query {
+                  field
+                }
+                """
+            )
+
+            field_details_list = collected.grouped_field_set.get("field")
+
+            assert field_details_list is not None
+            assert len(field_details_list) == 1

tests/execution/test_stream.py

@@ -13,6 +13,7 @@
     IncrementalStreamResult,
     StreamItemRecord,
     StreamRecord,
+    execute,
     experimental_execute_incrementally,
 )
 from graphql.language import DocumentNode, parse
@@ -2522,3 +2523,60 @@ async def iterable(_info):
         # is actually a generator that cannot do the cleanup in case of an athrow()
         await sleep(0)
         assert not finished
+
+    async def original_execute_function_throws_error_if_streamed_and_all_is_sync():
+        """Original execute function throws error if streamed and all is sync
+
+        Original execute function throws error if anything is streamed and everything
+        else is sync.
+        """
+        document = parse(
+            """
+            query {
+              scalarList
+              friendList @stream { name }
+            }
+            """
+        )
+
+        root_value = {
+            "scalarList": ["apple", "banana", "coconut"],
+            "friendList": [{"name": "Alice"}, {"name": "Bob"}],
+        }
+        with pytest.raises(GraphQLError) as exc_info:
+            await execute(schema, document, root_value)  # type: ignore
+
+        assert str(exc_info.value) == (
+            "Executing this GraphQL operation would unexpectedly produce"
+            " multiple payloads (due to @defer or @stream directive)"
+        )
+
+    async def original_execute_function_throws_error_if_streamed_and_not_all_is_sync():
+        """Original execute function throws error if streamed and not all is sync
+
+        Original execute function resolves to error if anything is streamed and
+        something else is async.
+        """
+        document = parse(
+            """
+            query {
+              scalarList
+              friendList @stream { name }
+            }
+            """
+        )
+
+        async def scalar_list(_info):
+            return ["apple", "banana", "coconut"]
+
+        root_value = {
+            "scalarList": scalar_list,
+            "friendList": [{"name": "Alice"}, {"name": "Bob"}],
+        }
+        with pytest.raises(GraphQLError) as exc_info:
+            await execute(schema, document, root_value)  # type: ignore
+
+        assert str(exc_info.value) == (
+            "Executing this GraphQL operation would unexpectedly produce"
+            " multiple payloads (due to @defer or @stream directive)"
+        )