diff --git a/docs-website/router/configuration.mdx b/docs-website/router/configuration.mdx index 99a855042c..0a5fbaa723 100644 --- a/docs-website/router/configuration.mdx +++ b/docs-website/router/configuration.mdx @@ -2058,6 +2058,8 @@ Configure the GraphQL Execution Engine of the Router. | ENGINE_VALIDATE_REQUIRED_EXTERNAL_FIELDS | validate_required_external_fields | | Validate nullable external "@requires" dependencies. When a subgraph entity fetch returns a null value with an error for a field set specified in the "@requires" directive, any following fetch that depends on it should not receive such an entity. | false | | ENGINE_SUBSCRIPTION_FETCH_TIMEOUT | subscription_fetch_timeout | | The maximum time a subscription fetch can take before it is considered timed out. | 30s | | ENGINE_ENABLE_DEFER | enable_defer | | Enables support for the `@defer` directive, allowing clients to defer parts of a query so that the initial response is returned faster and deferred fields are streamed incrementally. | false | +| ENGINE_ENABLE_MULTI_FETCH | enable_multi_fetch | | Merges entity fetches that target the same subgraph and execute at the same point in the query plan into a single request with aliased `_entities` fields. This reduces the number of requests sent to subgraphs. | false | +| ENGINE_ENABLE_SCHEDULE_FETCHES | enable_schedule_fetches | | Replaces the wave-based fetch execution with a dependency-aware fetch scheduler. Independent fetch chains progress as soon as their own dependencies complete, instead of waiting for the slowest fetch in each wave. | false | | ENGINE_RELAX_SUBGRAPH_OPERATION_FIELD_SELECTION_MERGING_NULLABILITY | relax_subgraph_operation_field_selection_merging_nullability | | Relaxes nullability validation for [field selection merging](/router/relaxed-field-selection-merging-nullability) across union member types. | false | | ENGINE_ALLOW_STRING_LITERALS_FOR_ENUMS | allow_string_literals_for_enums | | Accepts a [string literal where an enum value is expected](/router/string-literals-for-enums) if the string content matches one of the enum's values. This is a deviation from the GraphQL specification. | false | @@ -2106,6 +2108,8 @@ engine: disable_exposing_variables_content_on_validation_error: false enable_subgraph_fetch_operation_name: true subscription_fetch_timeout: 30s + enable_multi_fetch: false + enable_schedule_fetches: false relax_subgraph_operation_field_selection_merging_nullability: false allow_string_literals_for_enums: false ``` diff --git a/router-tests/go.mod b/router-tests/go.mod index cc9f3d6b00..f120748976 100644 --- a/router-tests/go.mod +++ b/router-tests/go.mod @@ -31,7 +31,7 @@ require ( github.com/wundergraph/cosmo/router v0.0.0-20260710155145-803a4bc06d92 github.com/wundergraph/cosmo/router-plugin v0.0.0-20250808194725-de123ba1c65e github.com/wundergraph/cosmo/speedtrap v0.0.0-00010101000000-000000000000 - github.com/wundergraph/graphql-go-tools/v2 v2.15.0 + github.com/wundergraph/graphql-go-tools/v2 v2.16.0 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/metric v1.44.0 diff --git a/router-tests/go.sum b/router-tests/go.sum index daf0bc5963..25513225dc 100644 --- a/router-tests/go.sum +++ b/router-tests/go.sum @@ -386,8 +386,8 @@ github.com/wundergraph/astjson v1.1.0 h1:xORDosrZ87zQFJwNGe/HIHXqzpdHOFmqWgykCLV github.com/wundergraph/astjson v1.1.0/go.mod h1:h12D/dxxnedtLzsKyBLK7/Oe4TAoGpRVC9nDpDrZSWw= github.com/wundergraph/go-arena v1.3.0 h1:n0ng5a1vbd8YGq1u3rMr0vPU5f6AZ1BXIiUhL1UIok8= github.com/wundergraph/go-arena v1.3.0/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw= -github.com/wundergraph/graphql-go-tools/v2 v2.15.0 h1:AG4l/QZrj1IBqxxPzz40r31fmKBopXQw+UjsgHYu8j4= -github.com/wundergraph/graphql-go-tools/v2 v2.15.0/go.mod h1:zREIKLmpjfNcGSubndaW/913r0Y8XbbYOXQeZFkwHdo= +github.com/wundergraph/graphql-go-tools/v2 v2.16.0 h1:zZ8XuHGfkWkMrqKvy2vc5u//Z94/t01leQhrTTbVOxo= +github.com/wundergraph/graphql-go-tools/v2 v2.16.0/go.mod h1:Q0DH6cCkFM/LAUT2ETlo6AMIZhUklZczF2I6uWK9HSA= github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 h1:FnBeRrxr7OU4VvAzt5X7s6266i6cSVkkFPS0TuXWbIg= github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= diff --git a/router-tests/operations/multi_fetches_scheduling_test.go b/router-tests/operations/multi_fetches_scheduling_test.go new file mode 100644 index 0000000000..d37388309f --- /dev/null +++ b/router-tests/operations/multi_fetches_scheduling_test.go @@ -0,0 +1,198 @@ +package integration + +import ( + "encoding/json" + "testing" + + "github.com/sebdah/goldie/v2" + "github.com/stretchr/testify/require" + + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router/pkg/config" +) + +func TestMultiFetch(t *testing.T) { + t.Parallel() + + // multiFetchQuery resolves Consultancy.lead from the employees subgraph, then needs + // two follow-up entity fetches: Employee.derivedMood and Consultancy.isLeadAvailable + const multiFetchQuery = `query Requires { + products { + __typename + ... on Consultancy { + lead { + __typename + id + derivedMood + } + isLeadAvailable + } + } + }` + const multiFetchExpectedResponse = `{"data":{"products":[{"__typename":"Consultancy","lead":{"__typename":"Employee","id":1,"derivedMood":"HAPPY"},"isLeadAvailable":false},{"__typename":"Cosmo"},{"__typename":"SDK"}]}}` + + t.Run("merges same-wave entity fetches to the same subgraph when enabled", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{ + ModifyEngineExecutionConfiguration: func(cfg *config.EngineExecutionConfiguration) { + cfg.EnableMultiFetch = true + }, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: multiFetchQuery}) + require.JSONEq(t, multiFetchExpectedResponse, res.Body) + + // Root fetch + one merged entity fetch. + require.EqualValues(t, 2, xEnv.SubgraphRequestCount.Employees.Load()) + + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Mood.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Availability.Load()) + require.EqualValues(t, 4, xEnv.SubgraphRequestCount.Global.Load()) + }) + }) + + t.Run("sends one request per entity fetch by default", func(t *testing.T) { + t.Parallel() + + testenv.Run(t, &testenv.Config{}, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: multiFetchQuery}) + // Merging must not change the result: same response as the merged case above. + require.JSONEq(t, multiFetchExpectedResponse, res.Body) + + // Root fetch + two separate entity fetches. + require.EqualValues(t, 3, xEnv.SubgraphRequestCount.Employees.Load()) + + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Mood.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Availability.Load()) + require.EqualValues(t, 5, xEnv.SubgraphRequestCount.Global.Load()) + }) + }) +} + +// queryPlanNode is a minimal mirror of the query plan JSON for structural assertions. +type queryPlanNode struct { + Kind string `json:"kind"` + Children []queryPlanNode `json:"children"` + Fetch *struct { + Kind string `json:"kind"` + SubgraphName string `json:"subgraphName"` + } `json:"fetch"` +} + +func (n *queryPlanNode) fetchKinds() []string { + var kinds []string + if n.Fetch != nil { + kinds = append(kinds, n.Fetch.Kind) + } + for i := range n.Children { + kinds = append(kinds, n.Children[i].fetchKinds()...) + } + return kinds +} + +// TestMultiFetchAndScheduleFetches verifies that enable_multi_fetch and enable_schedule_fetches +// can be enabled together and that both affect the generated query plan. +func TestMultiFetchAndScheduleFetches(t *testing.T) { + t.Parallel() + + newGoldie := func(t *testing.T) *goldie.Goldie { + return goldie.New( + t, + goldie.WithFixtureDir("testdata/fixtures/query_plans"), + goldie.WithNameSuffix(".json"), + goldie.WithDiffEngine(goldie.ClassicDiff), + ) + } + bothFlags := func(cfg *config.EngineExecutionConfiguration) { + cfg.EnableMultiFetch = true + cfg.EnableScheduleFetches = true + cfg.Debug.AlwaysIncludeQueryPlan = true + } + + queryPlanFromBody := func(t *testing.T, body string) queryPlanNode { + t.Helper() + var resp struct { + Extensions struct { + QueryPlan queryPlanNode `json:"queryPlan"` + } `json:"extensions"` + } + require.NoError(t, json.Unmarshal([]byte(body), &resp)) + return resp.Extensions.QueryPlan + } + + t.Run("same-subgraph entity fetches merge into one MultiEntity fetch", func(t *testing.T) { + t.Parallel() + + // Both employees entity fetches (derivedMood, isLeadAvailable requires) + // execute at the same point of the plan and merge. + const query = `query Requires { + products { + __typename + ... on Consultancy { + lead { + __typename + id + derivedMood + } + isLeadAvailable + } + } + }` + + testenv.Run(t, &testenv.Config{ + ModifyEngineExecutionConfiguration: bothFlags, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: query}) + newGoldie(t).Assert(t, "response_with_query_plan_multi_fetch_schedule", indentedJSON(res.Body)) + + plan := queryPlanFromBody(t, res.Body) + require.Contains(t, plan.fetchKinds(), "MultiEntity") + + // root fetch + 1 merged entity fetch to employees + require.EqualValues(t, 2, xEnv.SubgraphRequestCount.Employees.Load()) + + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Mood.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Availability.Load()) + require.EqualValues(t, 4, xEnv.SubgraphRequestCount.Global.Load()) + }) + }) + + t.Run("independent chains are scheduled without wave barriers", func(t *testing.T) { + t.Parallel() + + // Two independent chains: employee -> mood -> employees (derivedMood requires currentMood) + // and findEmployees (family) -> availability. + // The legacy organizer produces Sequence(Parallel, Parallel, ...) waves; + // the scheduler runs both chains side by side under a Parallel root. + const query = `query IndependentChains { + employee(id: 1) { + id + derivedMood + } + findEmployees(criteria: { nationality: GERMAN }) { + id + isAvailable + } + }` + + testenv.Run(t, &testenv.Config{ + ModifyEngineExecutionConfiguration: bothFlags, + }, func(t *testing.T, xEnv *testenv.Environment) { + res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: query}) + newGoldie(t).Assert(t, "response_with_query_plan_schedule_chains", indentedJSON(res.Body)) + + plan := queryPlanFromBody(t, res.Body) + require.Equal(t, "Parallel", plan.Kind, "scheduled plan must run independent chains in parallel") + require.Len(t, plan.Children, 2) + for _, chain := range plan.Children { + require.Equal(t, "Sequence", chain.Kind) + } + + require.EqualValues(t, 2, xEnv.SubgraphRequestCount.Employees.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Family.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Mood.Load()) + require.EqualValues(t, 1, xEnv.SubgraphRequestCount.Availability.Load()) + require.EqualValues(t, 5, xEnv.SubgraphRequestCount.Global.Load()) + }) + }) +} diff --git a/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_multi_fetch_schedule.json b/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_multi_fetch_schedule.json new file mode 100644 index 0000000000..f6bd05a044 --- /dev/null +++ b/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_multi_fetch_schedule.json @@ -0,0 +1,308 @@ +{ + "data": { + "products": [ + { + "__typename": "Consultancy", + "lead": { + "__typename": "Employee", + "id": 1, + "derivedMood": "HAPPY" + }, + "isLeadAvailable": false + }, + { + "__typename": "Cosmo" + }, + { + "__typename": "SDK" + } + ] + }, + "extensions": { + "queryPlan": { + "version": "1", + "kind": "Sequence", + "children": [ + { + "kind": "Single", + "fetch": { + "kind": "Single", + "subgraphName": "employees", + "subgraphId": "0", + "fetchId": 0, + "query": "{\n products {\n __typename\n ... on Consultancy {\n lead {\n __typename\n id\n }\n __typename\n upc\n }\n }\n}" + } + }, + { + "kind": "Parallel", + "children": [ + { + "kind": "Single", + "fetch": { + "kind": "BatchEntity", + "path": "products.@.lead", + "subgraphName": "availability", + "subgraphId": "5", + "fetchId": 1, + "dependsOnFetchIds": [ + 0 + ], + "representations": [ + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ], + "query": "query($representations: [_Any!]!){\n _entities(representations: $representations){\n ... on Employee {\n __typename\n isAvailable\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "isAvailable" + }, + "isUserRequested": false, + "dependsOn": [ + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + } + ] + } + }, + { + "kind": "Single", + "fetch": { + "kind": "BatchEntity", + "path": "products.@.lead", + "subgraphName": "mood", + "subgraphId": "6", + "fetchId": 2, + "dependsOnFetchIds": [ + 0 + ], + "representations": [ + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ], + "query": "query($representations: [_Any!]!){\n _entities(representations: $representations){\n ... on Employee {\n __typename\n currentMood\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "currentMood" + }, + "isUserRequested": false, + "dependsOn": [ + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + } + ] + } + } + ] + }, + { + "kind": "Single", + "fetch": { + "kind": "MultiEntity", + "subgraphName": "employees", + "subgraphId": "0", + "fetchId": 3, + "dependsOnFetchIds": [ + 0, + 2, + 1 + ], + "representations": [ + { + "kind": "@requires", + "typeName": "Employee", + "fieldName": "derivedMood", + "fragment": "fragment Requires_for_derivedMood on Employee {\n currentMood\n}" + }, + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + }, + { + "kind": "@requires", + "typeName": "Consultancy", + "fieldName": "isLeadAvailable", + "fragment": "fragment Requires_for_isLeadAvailable on Consultancy {\n lead {\n isAvailable\n }\n}" + }, + { + "kind": "@key", + "typeName": "Consultancy", + "fragment": "fragment Key on Consultancy {\n __typename\n upc\n}" + } + ], + "query": "query($representations_f1: [_Any!]!, $includeF1: Boolean!, $representations_f2: [_Any!]!, $includeF2: Boolean!){\n f1: _entities(representations: $representations_f1)@include(if: $includeF1) {\n ... on Employee {\n __typename\n derivedMood\n }\n }\n f2: _entities(representations: $representations_f2)@include(if: $includeF2) {\n ... on Consultancy {\n __typename\n isLeadAvailable\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "derivedMood" + }, + "isUserRequested": true, + "dependsOn": [ + { + "fetchId": 2, + "subgraph": "mood", + "coordinate": { + "typeName": "Employee", + "fieldName": "currentMood" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + }, + { + "coordinate": { + "typeName": "Consultancy", + "fieldName": "isLeadAvailable" + }, + "isUserRequested": true, + "dependsOn": [ + { + "fetchId": 3, + "subgraph": "employees", + "coordinate": { + "typeName": "Consultancy", + "fieldName": "lead" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 2, + "subgraph": "mood", + "coordinate": { + "typeName": "Consultancy", + "fieldName": "lead" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 1, + "subgraph": "availability", + "coordinate": { + "typeName": "Consultancy", + "fieldName": "lead" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Consultancy", + "fieldName": "lead" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 1, + "subgraph": "availability", + "coordinate": { + "typeName": "Employee", + "fieldName": "isAvailable" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Consultancy", + "fieldName": "upc" + }, + "isKey": true, + "isRequires": false + } + ] + } + ], + "mergedFetchIds": [ + 3, + 4 + ], + "entries": [ + { + "alias": "f1", + "path": "products.@.lead", + "representations": [ + { + "kind": "@requires", + "typeName": "Employee", + "fieldName": "derivedMood", + "fragment": "fragment Requires_for_derivedMood on Employee {\n currentMood\n}" + }, + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ] + }, + { + "alias": "f2", + "path": "products", + "representations": [ + { + "kind": "@requires", + "typeName": "Consultancy", + "fieldName": "isLeadAvailable", + "fragment": "fragment Requires_for_isLeadAvailable on Consultancy {\n lead {\n isAvailable\n }\n}" + }, + { + "kind": "@key", + "typeName": "Consultancy", + "fragment": "fragment Key on Consultancy {\n __typename\n upc\n}" + } + ] + } + ] + } + } + ], + "normalizedQuery": "query Requires {products {__typename ... on Consultancy {lead {__typename id derivedMood} isLeadAvailable}}}" + } + } +} \ No newline at end of file diff --git a/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_schedule_chains.json b/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_schedule_chains.json new file mode 100644 index 0000000000..7d9bb2162a --- /dev/null +++ b/router-tests/operations/testdata/fixtures/query_plans/response_with_query_plan_schedule_chains.json @@ -0,0 +1,208 @@ +{ + "data": { + "employee": { + "id": 1, + "derivedMood": "HAPPY" + }, + "findEmployees": [ + { + "id": 1, + "isAvailable": false + }, + { + "id": 2, + "isAvailable": false + }, + { + "id": 4, + "isAvailable": false + }, + { + "id": 11, + "isAvailable": false + } + ] + }, + "extensions": { + "queryPlan": { + "version": "1", + "kind": "Parallel", + "children": [ + { + "kind": "Sequence", + "children": [ + { + "kind": "Single", + "fetch": { + "kind": "Single", + "subgraphName": "employees", + "subgraphId": "0", + "fetchId": 0, + "query": "query($a: Int!){\n employee(id: $a){\n id\n __typename\n }\n}" + } + }, + { + "kind": "Single", + "fetch": { + "kind": "Entity", + "path": "employee", + "subgraphName": "mood", + "subgraphId": "6", + "fetchId": 1, + "dependsOnFetchIds": [ + 0 + ], + "representations": [ + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ], + "query": "query($representations: [_Any!]!){\n _entities(representations: $representations){\n ... on Employee {\n __typename\n currentMood\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "currentMood" + }, + "isUserRequested": false, + "dependsOn": [ + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + } + ] + } + }, + { + "kind": "Single", + "fetch": { + "kind": "Entity", + "path": "employee", + "subgraphName": "employees", + "subgraphId": "0", + "fetchId": 4, + "dependsOnFetchIds": [ + 0, + 1 + ], + "representations": [ + { + "kind": "@requires", + "typeName": "Employee", + "fieldName": "derivedMood", + "fragment": "fragment Requires_for_derivedMood on Employee {\n currentMood\n}" + }, + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ], + "query": "query($representations: [_Any!]!){\n _entities(representations: $representations){\n ... on Employee {\n __typename\n derivedMood\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "derivedMood" + }, + "isUserRequested": true, + "dependsOn": [ + { + "fetchId": 1, + "subgraph": "mood", + "coordinate": { + "typeName": "Employee", + "fieldName": "currentMood" + }, + "isKey": false, + "isRequires": true + }, + { + "fetchId": 0, + "subgraph": "employees", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + } + ] + } + } + ] + }, + { + "kind": "Sequence", + "children": [ + { + "kind": "Single", + "fetch": { + "kind": "Single", + "subgraphName": "family", + "subgraphId": "1", + "fetchId": 2, + "query": "query($b: SearchInput){\n findEmployees(criteria: $b){\n id\n __typename\n }\n}" + } + }, + { + "kind": "Single", + "fetch": { + "kind": "BatchEntity", + "path": "findEmployees", + "subgraphName": "availability", + "subgraphId": "5", + "fetchId": 3, + "dependsOnFetchIds": [ + 2 + ], + "representations": [ + { + "kind": "@key", + "typeName": "Employee", + "fragment": "fragment Key on Employee {\n __typename\n id\n}" + } + ], + "query": "query($representations: [_Any!]!){\n _entities(representations: $representations){\n ... on Employee {\n __typename\n isAvailable\n }\n }\n}", + "dependencies": [ + { + "coordinate": { + "typeName": "Employee", + "fieldName": "isAvailable" + }, + "isUserRequested": true, + "dependsOn": [ + { + "fetchId": 2, + "subgraph": "family", + "coordinate": { + "typeName": "Employee", + "fieldName": "id" + }, + "isKey": true, + "isRequires": false + } + ] + } + ] + } + } + ] + } + ], + "normalizedQuery": "query IndependentChains($a: Int!, $b: SearchInput){employee(id: $a){id derivedMood} findEmployees(criteria: $b){id isAvailable}}" + } + } +} \ No newline at end of file diff --git a/router/core/executor.go b/router/core/executor.go index 6c04b96b47..c15bf12288 100644 --- a/router/core/executor.go +++ b/router/core/executor.go @@ -18,6 +18,7 @@ import ( "github.com/wundergraph/graphql-go-tools/v2/pkg/asttransform" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/introspection_datasource" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/plan" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/postprocess" "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" "github.com/wundergraph/graphql-go-tools/v2/pkg/operationreport" ) @@ -49,6 +50,9 @@ type Executor struct { Resolver *resolve.Resolver RenameTypeNames []resolve.RenameTypeName TrackUsageInfo bool + // PostprocessorOptions configure the plan postprocessor from the engine + // execution configuration (multi-fetch merging, fetch scheduling). + PostprocessorOptions []postprocess.ProcessorOption } type ExecutorBuildOptions struct { @@ -209,13 +213,22 @@ func (b *ExecutorConfigurationBuilder) Build(ctx context.Context, opts *Executor } } + var postprocessorOptions []postprocess.ProcessorOption + if opts.RouterEngineConfig.Execution.EnableMultiFetch { + postprocessorOptions = append(postprocessorOptions, postprocess.EnableMultiFetch()) + } + if opts.RouterEngineConfig.Execution.EnableScheduleFetches { + postprocessorOptions = append(postprocessorOptions, postprocess.EnableScheduleFetches()) + } + return &Executor{ - PlanConfig: *planConfig, - ClientSchema: clientSchemaDefinition, - RouterSchema: &routerSchemaDefinition, - Resolver: resolver, - RenameTypeNames: renameTypeNames, - TrackUsageInfo: b.trackUsageInfo, + PlanConfig: *planConfig, + ClientSchema: clientSchemaDefinition, + RouterSchema: &routerSchemaDefinition, + Resolver: resolver, + RenameTypeNames: renameTypeNames, + TrackUsageInfo: b.trackUsageInfo, + PostprocessorOptions: postprocessorOptions, }, providers, nil } diff --git a/router/core/factoryresolver.go b/router/core/factoryresolver.go index 1fe79b6bfa..bde74efe70 100644 --- a/router/core/factoryresolver.go +++ b/router/core/factoryresolver.go @@ -311,6 +311,9 @@ func (l *Loader) Load(engineConfig *nodev1.EngineConfiguration, subgraphs []*nod var outConfig plan.Configuration // attach field usage information to the plan outConfig.DefaultFlushIntervalMillis = engineConfig.DefaultFlushInterval + // EnableMultiFetch makes the planner record the subgraph operation artifacts + // the postprocessor's multi-fetch merge stage consumes. + outConfig.EnableMultiFetch = routerEngineConfig.Execution.EnableMultiFetch for _, configuration := range engineConfig.FieldConfigurations { var args []plan.ArgumentConfiguration for _, argumentConfiguration := range configuration.ArgumentsConfiguration { diff --git a/router/core/operation_planner.go b/router/core/operation_planner.go index a4191ff550..ecc595eeb9 100644 --- a/router/core/operation_planner.go +++ b/router/core/operation_planner.go @@ -2,6 +2,7 @@ package core import ( "errors" + "slices" "strconv" "time" @@ -98,7 +99,10 @@ func (p *OperationPlanner) planOperation(content string, name string, includeQue } // postprocess query plan to get its final state - post := postprocess.NewProcessor(postprocess.CollectDataSourceInfo()) + post := postprocess.NewProcessor(append( + slices.Clone(p.executor.PostprocessorOptions), + postprocess.CollectDataSourceInfo(), + )...) post.Process(preparedPlan) return &planWithMetaData{ diff --git a/router/go.mod b/router/go.mod index 5ba70bfe60..6e6c739168 100644 --- a/router/go.mod +++ b/router/go.mod @@ -31,7 +31,7 @@ require ( github.com/tidwall/gjson v1.18.0 github.com/tidwall/sjson v1.2.5 github.com/twmb/franz-go v1.16.1 - github.com/wundergraph/graphql-go-tools/v2 v2.15.0 + github.com/wundergraph/graphql-go-tools/v2 v2.16.0 // Do not upgrade, it renames attributes we rely on go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 go.opentelemetry.io/contrib/propagators/b3 v1.44.0 diff --git a/router/go.sum b/router/go.sum index 98af7c6676..a3c79cf627 100644 --- a/router/go.sum +++ b/router/go.sum @@ -334,8 +334,8 @@ github.com/wundergraph/astjson v1.1.0 h1:xORDosrZ87zQFJwNGe/HIHXqzpdHOFmqWgykCLV github.com/wundergraph/astjson v1.1.0/go.mod h1:h12D/dxxnedtLzsKyBLK7/Oe4TAoGpRVC9nDpDrZSWw= github.com/wundergraph/go-arena v1.3.0 h1:n0ng5a1vbd8YGq1u3rMr0vPU5f6AZ1BXIiUhL1UIok8= github.com/wundergraph/go-arena v1.3.0/go.mod h1:ROOysEHWJjLQ8FSfNxZCziagb7Qw2nXY3/vgKRh7eWw= -github.com/wundergraph/graphql-go-tools/v2 v2.15.0 h1:AG4l/QZrj1IBqxxPzz40r31fmKBopXQw+UjsgHYu8j4= -github.com/wundergraph/graphql-go-tools/v2 v2.15.0/go.mod h1:zREIKLmpjfNcGSubndaW/913r0Y8XbbYOXQeZFkwHdo= +github.com/wundergraph/graphql-go-tools/v2 v2.16.0 h1:zZ8XuHGfkWkMrqKvy2vc5u//Z94/t01leQhrTTbVOxo= +github.com/wundergraph/graphql-go-tools/v2 v2.16.0/go.mod h1:Q0DH6cCkFM/LAUT2ETlo6AMIZhUklZczF2I6uWK9HSA= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go index 3b905316e5..0b1489a051 100644 --- a/router/pkg/config/config.go +++ b/router/pkg/config/config.go @@ -9,10 +9,11 @@ import ( "github.com/caarlos0/env/v11" "github.com/goccy/go-yaml" + "go.uber.org/zap/zapcore" + "github.com/wundergraph/cosmo/router/internal/unique" "github.com/wundergraph/cosmo/router/internal/yamlmerge" "github.com/wundergraph/cosmo/router/pkg/otel/otelconfig" - "go.uber.org/zap/zapcore" ) const ( @@ -479,8 +480,9 @@ type EngineExecutionConfiguration struct { // Deprecated: EnableExecutionPlanCacheResponseHeader is deprecated, use EngineDebugConfiguration.EnableCacheResponseHeaders instead. EnableExecutionPlanCacheResponseHeader bool `envDefault:"false" env:"ENGINE_ENABLE_EXECUTION_PLAN_CACHE_RESPONSE_HEADER" yaml:"enable_execution_plan_cache_response_header"` - MaxConcurrentResolvers int `envDefault:"1024" env:"ENGINE_MAX_CONCURRENT_RESOLVERS" yaml:"max_concurrent_resolvers,omitempty"` - EnableNetPoll bool `envDefault:"true" env:"ENGINE_ENABLE_NET_POLL" yaml:"enable_net_poll"` + MaxConcurrentResolvers int `envDefault:"1024" env:"ENGINE_MAX_CONCURRENT_RESOLVERS" yaml:"max_concurrent_resolvers,omitempty"` + EnableNetPoll bool `envDefault:"true" env:"ENGINE_ENABLE_NET_POLL" yaml:"enable_net_poll"` + ExecutionPlanCacheSize int64 `envDefault:"1024" env:"ENGINE_EXECUTION_PLAN_CACHE_SIZE" yaml:"execution_plan_cache_size,omitempty"` SlowPlanCacheSize int64 `envDefault:"300" env:"ENGINE_SLOW_PLAN_CACHE_SIZE" yaml:"slow_plan_cache_size,omitempty"` SlowPlanCacheThreshold time.Duration `envDefault:"100ms" env:"ENGINE_SLOW_PLAN_CACHE_THRESHOLD" yaml:"slow_plan_cache_threshold,omitempty"` @@ -500,6 +502,13 @@ type EngineExecutionConfiguration struct { SubscriptionFetchTimeout time.Duration `envDefault:"30s" env:"ENGINE_SUBSCRIPTION_FETCH_TIMEOUT" yaml:"subscription_fetch_timeout,omitempty"` EnableDefer bool `envDefault:"false" env:"ENGINE_ENABLE_DEFER" yaml:"enable_defer"` + // EnableMultiFetch merges entity fetches to the same subgraph that execute + // in the same wave into a single batched request with aliased _entities fields. + EnableMultiFetch bool `envDefault:"false" env:"ENGINE_ENABLE_MULTI_FETCH" yaml:"enable_multi_fetch"` + // EnableScheduleFetches replaces the legacy wave-based fetch organizers with the + // dependency-aware fetch scheduler (component-split, chain-inlined execution trees). + EnableScheduleFetches bool `envDefault:"false" env:"ENGINE_ENABLE_SCHEDULE_FETCHES" yaml:"enable_schedule_fetches"` + // Server-side WebSocket handler options (router accepting client connections) WebSocketServerReadTimeout time.Duration `envDefault:"5s" env:"ENGINE_WEBSOCKET_SERVER_READ_TIMEOUT" yaml:"websocket_server_read_timeout,omitempty"` WebSocketServerWriteTimeout time.Duration `envDefault:"10s" env:"ENGINE_WEBSOCKET_SERVER_WRITE_TIMEOUT" yaml:"websocket_server_write_timeout,omitempty"` diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json index 92f83949f4..ac6bf7d5ac 100644 --- a/router/pkg/config/config.schema.json +++ b/router/pkg/config/config.schema.json @@ -3973,6 +3973,16 @@ "description": "Enables the more efficient poll implementation for all WebSocket implementations (client, server) of the router. This is only available on Linux and MacOS. On Windows or when the host system is limited, the default synchronous implementation is used.", "default": true }, + "enable_multi_fetch": { + "type": "boolean", + "description": "Enables merging entity fetches to the same subgraph that execute in the same wave into a single batched request with aliased _entities fields. Merging reduces the number of subgraph requests.", + "default": false + }, + "enable_schedule_fetches": { + "type": "boolean", + "description": "Enables the dependency-aware fetch scheduler, which organizes fetches into component-split, chain-inlined execution trees instead of the legacy wave-based structure.", + "default": false + }, "websocket_client_write_timeout": { "type": "string", "format": "go-duration", diff --git a/router/pkg/config/testdata/config_defaults.json b/router/pkg/config/testdata/config_defaults.json index 6df104d98a..576760616c 100644 --- a/router/pkg/config/testdata/config_defaults.json +++ b/router/pkg/config/testdata/config_defaults.json @@ -517,6 +517,8 @@ "EnableRequireFetchReasons": false, "SubscriptionFetchTimeout": 30000000000, "EnableDefer": false, + "EnableMultiFetch": false, + "EnableScheduleFetches": false, "WebSocketServerReadTimeout": 5000000000, "WebSocketServerWriteTimeout": 10000000000, "WebSocketServerPollTimeout": 1000000000, diff --git a/router/pkg/config/testdata/config_full.json b/router/pkg/config/testdata/config_full.json index 6d79d4ad91..1539deedab 100644 --- a/router/pkg/config/testdata/config_full.json +++ b/router/pkg/config/testdata/config_full.json @@ -985,6 +985,8 @@ "EnableRequireFetchReasons": false, "SubscriptionFetchTimeout": 30000000000, "EnableDefer": false, + "EnableMultiFetch": false, + "EnableScheduleFetches": false, "WebSocketServerReadTimeout": 5000000000, "WebSocketServerWriteTimeout": 10000000000, "WebSocketServerPollTimeout": 1000000000,