[Fusion] Add support for result from different subgraph on interfaces

src/HotChocolate/Fusion/src/Core/Planning/Pipeline/ExecutionStepDiscoveryMiddleware.cs

@@ -435,14 +435,44 @@ private void CollectNestedSelections(
                 subgraph.Add(executionStep.SubgraphName);
             }
 
-            backlog.Enqueue(
-                new BacklogItem(
-                    parentSelection,
-                    selectionSetPath,
-                    declaringType,
-                    leftovers,
-                    preferBatching));
+            TryEnqueueBacklogItem(
+                backlog,
+                parentSelection,
+                selectionSetPath,
+                declaringType,
+                leftovers,
+                preferBatching
+            );
+        }
+    }
+
+    private static void TryEnqueueBacklogItem(
+        Queue<BacklogItem> backlog, 
+        ISelection parentSelection, 
+        SelectionPath? selectionSetPath, 
+        ObjectTypeMetadata declaringType, 
+        List<ISelection> leftovers, 
+        bool preferBatching)
+    {
+        foreach (var item in backlog)
+        {
+            if ((item.SelectionPath?.Equals(selectionSetPath) ?? selectionSetPath is null) &&
+                item.DeclaringTypeMetadata == declaringType &&
+                item.PreferBatching == preferBatching &&
+                item.Selections.Count == leftovers.Count &&
+                item.Selections.SequenceEqual(leftovers))
+            {
+                return;
+            }
         }
+
+        backlog.Enqueue(
+            new BacklogItem(
+                parentSelection,
+                selectionSetPath,
+                declaringType,
+                leftovers,
+                preferBatching));
     }
 
     private static SelectionPath? CreateSelectionPath(SelectionPath? rootPath, List<ISelection> pathSegments)

src/HotChocolate/Fusion/test/Core.Tests/RequestPlannerTests.cs

@@ -2794,6 +2794,71 @@ type Query {
         await snapshot.MatchMarkdownAsync();
     }
 
+    [Fact]
+    public async Task Subgraph_Requested_On_Interface_Called_Only_Once()
+    {
+        // arrange
+        var subgraphA = await TestSubgraph.CreateAsync(
+            """
+            schema {
+              query: Query
+            }
+
+            type Query {
+              books: [Book!]!
+            }
+
+            interface Book {
+                author: Author!
+            }
+
+            type FunnyBook implements Book {
+                author: Author!
+            }
+
+            type ScaryBook implements Book {
+                author: Author!
+            }
+
+            type Author {
+              id: Int!
+            }
+            """
+        );
+
+        var subgraphB = await TestSubgraph.CreateAsync(
+            """
+            schema {
+              query: Query
+            }
+
+            type Query {
+              authorById(id: [Int!]!): [Author!]!
+            }
+
+            type Author {
+              id: Int!
+              name: String!
+            }
+            """
+        );
+
+        using var subgraphs = new TestSubgraphCollection(output, [subgraphA, subgraphB]);
+        var fusionGraph = await subgraphs.GetFusionGraphAsync();
+
+        // act
+        var result = await CreateQueryPlanAsync(
+            fusionGraph,
+            //"query { subgraph2Foo { name } }");
+            "query { books { author { name } } }");
+
+        // assert
+        var snapshot = new Snapshot();
+        snapshot.Add(result.UserRequest, nameof(result.UserRequest));
+        snapshot.Add(result.QueryPlan, nameof(result.QueryPlan));
+        await snapshot.MatchMarkdownAsync();
+    }
+
     private static async Task<(DocumentNode UserRequest, Execution.Nodes.QueryPlan QueryPlan)> CreateQueryPlanAsync(
         Skimmed.SchemaDefinition fusionGraph,
         [StringSyntax("graphql")] string query)

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/RequestPlannerTests.Subgraph_Requested_On_Interface_Called_Only_Once.md

@@ -0,0 +1,67 @@
+# Subgraph_Requested_On_Interface_Called_Only_Once
+
+## UserRequest
+
+```graphql
+{
+  books {
+    author {
+      name
+    }
+  }
+}
+```
+
+## QueryPlan
+
+```json
+{
+  "document": "{ books { author { name } } }",
+  "rootNode": {
+    "type": "Sequence",
+    "nodes": [
+      {
+        "type": "Resolve",
+        "subgraph": "Subgraph_1",
+        "document": "query fetch_books_1 { books { __typename ... on ScaryBook { author { __fusion_exports__1: id } } ... on FunnyBook { author { __fusion_exports__1: id } } } }",
+        "selectionSetId": 0,
+        "provides": [
+          {
+            "variable": "__fusion_exports__1"
+          }
+        ]
+      },
+      {
+        "type": "Compose",
+        "selectionSetIds": [
+          0
+        ]
+      },
+      {
+        "type": "ResolveByKeyBatch",
+        "subgraph": "Subgraph_2",
+        "document": "query fetch_books_2($__fusion_exports__1: [Int!]!) { authorById(id: $__fusion_exports__1) { name __fusion_exports__1: id } }",
+        "selectionSetId": 3,
+        "path": [
+          "authorById"
+        ],
+        "requires": [
+          {
+            "variable": "__fusion_exports__1"
+          }
+        ]
+      },
+      {
+        "type": "Compose",
+        "selectionSetIds": [
+          3
+        ]
+      }
+    ]
+  },
+  "state": {
+    "__fusion_exports__1": "Author_id"
+  }
+}
+```
+