1use std::cell::Cell;
2use std::num::NonZeroU32;
3use std::ops::ControlFlow;
4use std::sync::Arc;
5
6use apollo_compiler::ExecutableDocument;
7use apollo_compiler::Name;
8use apollo_compiler::collections::IndexMap;
9use apollo_compiler::collections::IndexSet;
10use apollo_compiler::validation::Valid;
11use itertools::Itertools;
12use petgraph::visit::EdgeRef;
13use serde::Deserialize;
14use serde::Serialize;
15use tracing::trace;
16
17use super::ConditionNode;
18use super::QueryPlanCost;
19use super::fetch_dependency_graph::FetchIdGenerator;
20use crate::ApiSchemaOptions;
21use crate::Supergraph;
22use crate::bail;
23use crate::error::FederationError;
24use crate::error::SingleFederationError;
25use crate::internal_error;
26use crate::operation::NormalizedDefer;
27use crate::operation::Operation;
28use crate::operation::SelectionSet;
29use crate::operation::normalize_operation;
30use crate::query_graph::OverrideConditions;
31use crate::query_graph::QueryGraph;
32use crate::query_graph::QueryGraphNodeType;
33use crate::query_graph::build_federated_query_graph;
34use crate::query_graph::condition_resolver::ConditionResolverCache;
35use crate::query_graph::path_tree::OpPathTree;
36use crate::query_plan::PlanNode;
37use crate::query_plan::QueryPlan;
38use crate::query_plan::SequenceNode;
39use crate::query_plan::TopLevelPlanNode;
40use crate::query_plan::fetch_dependency_graph::FetchDependencyGraph;
41use crate::query_plan::fetch_dependency_graph::FetchDependencyGraphNodePath;
42use crate::query_plan::fetch_dependency_graph::compute_nodes_for_tree;
43use crate::query_plan::fetch_dependency_graph_processor::FetchDependencyGraphProcessor;
44use crate::query_plan::fetch_dependency_graph_processor::FetchDependencyGraphToCostProcessor;
45use crate::query_plan::fetch_dependency_graph_processor::FetchDependencyGraphToQueryPlanProcessor;
46use crate::query_plan::query_planning_traversal::BestQueryPlanInfo;
47use crate::query_plan::query_planning_traversal::QueryPlanningParameters;
48use crate::query_plan::query_planning_traversal::QueryPlanningTraversal;
49use crate::query_plan::query_planning_traversal::convert_type_from_subgraph;
50use crate::query_plan::query_planning_traversal::non_local_selections_estimation;
51use crate::schema::ValidFederationSchema;
52use crate::schema::position::AbstractTypeDefinitionPosition;
53use crate::schema::position::CompositeTypeDefinitionPosition;
54use crate::schema::position::InterfaceTypeDefinitionPosition;
55use crate::schema::position::ObjectTypeDefinitionPosition;
56use crate::schema::position::OutputTypeDefinitionPosition;
57use crate::schema::position::SchemaRootDefinitionKind;
58use crate::schema::position::TypeDefinitionPosition;
59use crate::utils::logging::snapshot;
60
61#[derive(Debug, Clone, Hash, Serialize)]
62pub struct QueryPlannerConfig {
63 pub generate_query_fragments: bool,
68
69 pub subgraph_graphql_validation: bool,
77
78 pub incremental_delivery: QueryPlanIncrementalDeliveryConfig,
84
85 pub debug: QueryPlannerDebugConfig,
89
90 pub type_conditioned_fetching: bool,
97}
98
99#[allow(clippy::derivable_impls)] impl Default for QueryPlannerConfig {
101 fn default() -> Self {
102 Self {
103 generate_query_fragments: false,
104 subgraph_graphql_validation: false,
105 incremental_delivery: Default::default(),
106 debug: Default::default(),
107 type_conditioned_fetching: false,
108 }
109 }
110}
111
112#[derive(Debug, Clone, Default, Hash, Serialize)]
113pub struct QueryPlanIncrementalDeliveryConfig {
114 #[serde(default)]
124 pub enable_defer: bool,
125}
126
127#[derive(Debug, Clone, Hash, Serialize)]
128pub struct QueryPlannerDebugConfig {
129 pub max_evaluated_plans: NonZeroU32,
147
148 pub paths_limit: Option<u32>,
161}
162
163impl Default for QueryPlannerDebugConfig {
164 fn default() -> Self {
165 Self {
166 max_evaluated_plans: NonZeroU32::new(10_000).unwrap(),
167 paths_limit: None,
168 }
169 }
170}
171
172#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
174pub struct QueryPlanningStatistics {
175 pub evaluated_plan_count: Cell<usize>,
176 pub evaluated_plan_paths: Cell<usize>,
177 #[serde(deserialize_with = "deserialize_f64_nullable")]
179 pub best_plan_cost: f64,
180}
181
182fn deserialize_f64_nullable<'de, D>(deserializer: D) -> Result<f64, D::Error>
185where
186 D: serde::de::Deserializer<'de>,
187{
188 let opt = Option::<f64>::deserialize(deserializer)?;
190 Ok(opt.unwrap_or(f64::NAN))
192}
193
194#[derive(Clone)]
195pub struct QueryPlanOptions<'a> {
196 pub override_conditions: Vec<String>,
203 pub check_for_cooperative_cancellation: Option<&'a dyn Fn() -> ControlFlow<()>>,
212 pub non_local_selections_limit_enabled: bool,
215 pub disabled_subgraph_names: IndexSet<String>,
220}
221
222impl Default for QueryPlanOptions<'_> {
223 fn default() -> Self {
224 Self {
225 override_conditions: Vec::new(),
226 check_for_cooperative_cancellation: None,
227 non_local_selections_limit_enabled: true,
228 disabled_subgraph_names: Default::default(),
229 }
230 }
231}
232
233impl std::fmt::Debug for QueryPlanOptions<'_> {
234 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
235 f.debug_struct("QueryPlanOptions")
236 .field("override_conditions", &self.override_conditions)
237 .field(
238 "check_for_cooperative_cancellation",
239 if self.check_for_cooperative_cancellation.is_some() {
240 &"Some(...)"
241 } else {
242 &"None"
243 },
244 )
245 .field(
246 "non_local_selections_limit_enabled",
247 &self.non_local_selections_limit_enabled,
248 )
249 .finish()
250 }
251}
252
253pub struct QueryPlanner {
254 config: QueryPlannerConfig,
255 federated_query_graph: Arc<QueryGraph>,
256 supergraph_schema: ValidFederationSchema,
257 api_schema: ValidFederationSchema,
258 interface_types_with_interface_objects: IndexSet<InterfaceTypeDefinitionPosition>,
261 abstract_types_with_inconsistent_runtime_types: IndexSet<Name>,
266}
267
268impl QueryPlanner {
269 #[cfg_attr(
270 feature = "snapshot_tracing",
271 tracing::instrument(level = "trace", skip_all, name = "QueryPlanner::new")
272 )]
273 pub fn new(
274 supergraph: &Supergraph,
275 config: QueryPlannerConfig,
276 ) -> Result<Self, FederationError> {
277 let supergraph_schema = supergraph.schema.clone();
278 let api_schema = supergraph.to_api_schema(ApiSchemaOptions {
279 include_defer: config.incremental_delivery.enable_defer,
280 ..Default::default()
281 })?;
282 let query_graph = build_federated_query_graph(
283 supergraph_schema.clone(),
284 api_schema.clone(),
285 Some(true),
286 Some(true),
287 )?;
288
289 let interface_types_with_interface_objects = supergraph
290 .schema
291 .get_types()
292 .filter_map(|position| match position {
293 TypeDefinitionPosition::Interface(interface_position) => Some(interface_position),
294 _ => None,
295 })
296 .map(|position| {
297 let is_interface_object = query_graph
298 .subgraphs()
299 .map(|(_name, schema)| {
300 let Some(position) = schema.try_get_type(&position.type_name) else {
301 return Ok(false);
302 };
303 schema.is_interface_object_type(position)
304 })
305 .process_results(|mut iter| iter.any(|b| b))?;
306 Ok::<_, FederationError>((position, is_interface_object))
307 })
308 .process_results(|iter| {
309 iter.flat_map(|(position, is_interface_object)| {
310 if is_interface_object {
311 Some(position)
312 } else {
313 None
314 }
315 })
316 .collect::<IndexSet<_>>()
317 })?;
318
319 let is_inconsistent = |position: AbstractTypeDefinitionPosition| {
320 let mut sources = query_graph.subgraphs().filter_map(|(_name, subgraph)| {
321 match subgraph.try_get_type(position.type_name())? {
322 TypeDefinitionPosition::Object(_) => None,
327 TypeDefinitionPosition::Interface(interface) => Some(
328 subgraph
329 .referencers()
330 .get_interface_type(&interface.type_name)
331 .ok()?
332 .object_types
333 .clone(),
334 ),
335 TypeDefinitionPosition::Union(union_) => Some(
336 union_
337 .try_get(subgraph.schema())?
338 .members
339 .iter()
340 .map(|member| ObjectTypeDefinitionPosition::new(member.name.clone()))
341 .collect(),
342 ),
343 _ => None,
344 }
345 });
346
347 let Some(expected_runtimes) = sources.next() else {
348 return false;
349 };
350 !sources.all(|runtimes| runtimes == expected_runtimes)
351 };
352
353 let abstract_types_with_inconsistent_runtime_types = supergraph
354 .schema
355 .get_types()
356 .filter_map(|position| AbstractTypeDefinitionPosition::try_from(position).ok())
357 .filter(|position| is_inconsistent(position.clone()))
358 .map(|position| position.type_name().clone())
359 .collect::<IndexSet<_>>();
360
361 Ok(Self {
362 config,
363 federated_query_graph: Arc::new(query_graph),
364 supergraph_schema,
365 api_schema,
366 interface_types_with_interface_objects,
367 abstract_types_with_inconsistent_runtime_types,
368 })
369 }
370
371 pub fn subgraph_schemas(&self) -> &IndexMap<Arc<str>, ValidFederationSchema> {
372 self.federated_query_graph.subgraph_schemas()
373 }
374
375 #[cfg_attr(
377 feature = "snapshot_tracing",
378 tracing::instrument(level = "trace", skip_all, name = "QueryPlanner::build_query_plan")
379 )]
380 pub fn build_query_plan(
381 &self,
382 document: &Valid<ExecutableDocument>,
383 operation_name: Option<Name>,
384 options: QueryPlanOptions,
385 ) -> Result<QueryPlan, FederationError> {
386 let operation = document
387 .operations
388 .get(operation_name.as_ref().map(|name| name.as_str()))
389 .map_err(|_| {
390 if operation_name.is_some() {
391 SingleFederationError::UnknownOperation
392 } else {
393 SingleFederationError::OperationNameNotProvided
394 }
395 })?;
396 if operation.selection_set.is_empty() {
397 crate::bail!("Invalid operation: empty selection set")
399 }
400
401 let is_subscription = operation.is_subscription();
402
403 let statistics = QueryPlanningStatistics::default();
404
405 let normalized_operation = normalize_operation(
406 operation,
407 &document.fragments,
408 &self.api_schema,
409 &self.interface_types_with_interface_objects,
410 &|| {
411 QueryPlanningParameters::check_cancellation_with(
412 &options.check_for_cooperative_cancellation,
413 )
414 },
415 )?;
416
417 let NormalizedDefer {
418 operation: normalized_operation,
419 assigned_defer_labels,
420 defer_conditions,
421 has_defers,
422 } = normalized_operation.with_normalized_defer()?;
423 if has_defers && is_subscription {
424 return Err(SingleFederationError::DeferredSubscriptionUnsupported.into());
425 }
426
427 if normalized_operation.selection_set.is_empty() {
428 return Ok(QueryPlan::default());
429 }
430
431 snapshot!(
432 "NormalizedOperation",
433 serde_json_bytes::json!({
434 "original": &operation.serialize().to_string(),
435 "normalized": &normalized_operation.to_string()
436 })
437 .to_string(),
438 "normalized operation"
439 );
440
441 let Some(root) = self
442 .federated_query_graph
443 .root_kinds_to_nodes()?
444 .get(&normalized_operation.root_kind)
445 else {
446 bail!(
447 "Shouldn't have a {0} operation if the subgraphs don't have a {0} root",
448 normalized_operation.root_kind
449 )
450 };
451
452 let operation_compression = if self.config.generate_query_fragments {
453 SubgraphOperationCompression::GenerateFragments
454 } else {
455 SubgraphOperationCompression::Disabled
456 };
457 let mut processor = FetchDependencyGraphToQueryPlanProcessor::new(
458 normalized_operation.variables.clone(),
459 normalized_operation.directives.clone(),
460 operation_compression,
461 operation.name.clone(),
462 assigned_defer_labels,
463 );
464 let mut parameters = QueryPlanningParameters {
465 supergraph_schema: self.supergraph_schema.clone(),
466 federated_query_graph: self.federated_query_graph.clone(),
467 operation: Arc::new(normalized_operation),
468 head: *root,
469 head_must_be_root: true,
472 statistics: &statistics,
473 abstract_types_with_inconsistent_runtime_types: self
474 .abstract_types_with_inconsistent_runtime_types
475 .clone()
476 .into(),
477 config: self.config.clone(),
478 override_conditions: OverrideConditions::new(
479 &self.federated_query_graph,
480 &IndexSet::from_iter(options.override_conditions),
481 ),
482 check_for_cooperative_cancellation: options.check_for_cooperative_cancellation,
483 fetch_id_generator: Arc::new(FetchIdGenerator::new()),
484 disabled_subgraphs: self
485 .federated_query_graph
486 .subgraphs()
487 .filter_map(|(subgraph, _)| {
488 if options.disabled_subgraph_names.contains(subgraph.as_ref()) {
489 Some(subgraph.clone())
490 } else {
491 None
492 }
493 })
494 .collect(),
495 };
496
497 let mut non_local_selection_state = options
498 .non_local_selections_limit_enabled
499 .then(non_local_selections_estimation::State::default);
500 let mut resolver_cache = ConditionResolverCache::new();
501 let (root_node, cost) = if !defer_conditions.is_empty() {
502 compute_plan_for_defer_conditionals(
503 &mut parameters,
504 &mut processor,
505 defer_conditions,
506 &mut non_local_selection_state,
507 &mut resolver_cache,
508 )
509 } else {
510 compute_plan_internal(
511 &mut parameters,
512 &mut processor,
513 has_defers,
514 &mut non_local_selection_state,
515 &mut resolver_cache,
516 )
517 }?;
518
519 let root_node = match root_node {
520 Some(PlanNode::Fetch(root_node)) if is_subscription => Some(
524 TopLevelPlanNode::Subscription(crate::query_plan::SubscriptionNode {
525 primary: root_node,
526 rest: None,
527 }),
528 ),
529 Some(PlanNode::Sequence(root_node)) if is_subscription => {
530 let Some((primary, rest)) = root_node.nodes.split_first() else {
531 bail!("Invalid query plan: Sequence must have at least one node");
533 };
534 let PlanNode::Fetch(primary) = primary.clone() else {
535 bail!("Invalid query plan: Primary node of a subscription is not a Fetch");
536 };
537 let rest = PlanNode::Sequence(SequenceNode {
538 nodes: rest.to_vec(),
539 });
540 Some(TopLevelPlanNode::Subscription(
541 crate::query_plan::SubscriptionNode {
542 primary,
543 rest: Some(Box::new(rest)),
544 },
545 ))
546 }
547 Some(node) if is_subscription => {
548 bail!(
549 "Invalid query plan for subscription: unexpected {} at root",
550 node.node_kind()
551 );
552 }
553 Some(PlanNode::Fetch(inner)) => Some(TopLevelPlanNode::Fetch(inner)),
554 Some(PlanNode::Sequence(inner)) => Some(TopLevelPlanNode::Sequence(inner)),
555 Some(PlanNode::Parallel(inner)) => Some(TopLevelPlanNode::Parallel(inner)),
556 Some(PlanNode::Flatten(inner)) => Some(TopLevelPlanNode::Flatten(inner)),
557 Some(PlanNode::Defer(inner)) => Some(TopLevelPlanNode::Defer(inner)),
558 Some(PlanNode::Condition(inner)) => Some(TopLevelPlanNode::Condition(inner)),
559 None => None,
560 };
561
562 let plan = QueryPlan {
563 node: root_node,
564 statistics: QueryPlanningStatistics {
565 best_plan_cost: cost,
566 ..statistics
567 },
568 };
569
570 snapshot!(
571 "QueryPlan",
572 plan.to_string(),
573 "QueryPlan from build_query_plan"
574 );
575 snapshot!(
576 plan.statistics,
577 "QueryPlanningStatistics from build_query_plan"
578 );
579
580 Ok(plan)
581 }
582
583 pub fn api_schema(&self) -> &ValidFederationSchema {
585 &self.api_schema
586 }
587
588 pub fn supergraph_schema(&self) -> &ValidFederationSchema {
589 &self.supergraph_schema
590 }
591
592 pub fn override_condition_labels(&self) -> &IndexSet<Arc<str>> {
593 self.federated_query_graph.override_condition_labels()
594 }
595}
596
597fn compute_root_serial_dependency_graph_for_mutation(
598 parameters: &QueryPlanningParameters,
599 has_defers: bool,
600 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
601 resolver_cache: &mut ConditionResolverCache,
602) -> Result<Vec<FetchDependencyGraph>, FederationError> {
603 let QueryPlanningParameters {
604 supergraph_schema,
605 federated_query_graph,
606 operation,
607 ..
608 } = parameters;
609 let root_type: Option<CompositeTypeDefinitionPosition> = if has_defers {
610 supergraph_schema
611 .schema()
612 .root_operation(operation.root_kind.into())
613 .and_then(|name| supergraph_schema.try_get_type(name))
614 .and_then(|ty| ty.try_into().ok())
615 } else {
616 None
617 };
618 let mut split_roots = operation.selection_set.clone().split_top_level_fields();
620 let mut digest = Vec::new();
621 let selection_set = split_roots
622 .next()
623 .ok_or_else(|| FederationError::internal("Empty top level fields"))?;
624 let BestQueryPlanInfo {
625 mut fetch_dependency_graph,
626 path_tree: mut prev_path,
627 ..
628 } = compute_root_parallel_best_plan_for_mutation(
629 parameters,
630 selection_set,
631 has_defers,
632 non_local_selection_state,
633 resolver_cache,
634 )?;
635 let mut prev_subgraph = only_root_subgraph(&fetch_dependency_graph)?;
636 for selection_set in split_roots {
637 let BestQueryPlanInfo {
638 fetch_dependency_graph: new_dep_graph,
639 path_tree: new_path,
640 ..
641 } = compute_root_parallel_best_plan_for_mutation(
642 parameters,
643 selection_set,
644 has_defers,
645 non_local_selection_state,
646 resolver_cache,
647 )?;
648 let new_subgraph = only_root_subgraph(&new_dep_graph)?;
649 if new_subgraph == prev_subgraph {
650 Arc::make_mut(&mut prev_path).extend(&new_path);
660 fetch_dependency_graph = FetchDependencyGraph::new(
661 supergraph_schema.clone(),
662 federated_query_graph.clone(),
663 root_type.clone(),
664 fetch_dependency_graph.fetch_id_generation.clone(),
665 );
666 compute_root_fetch_groups(
667 operation.root_kind,
668 federated_query_graph,
669 &mut fetch_dependency_graph,
670 &prev_path,
671 parameters.config.type_conditioned_fetching,
672 &|| parameters.check_cancellation(),
673 )?;
674 } else {
675 digest.push(std::mem::replace(
680 &mut fetch_dependency_graph,
681 new_dep_graph,
682 ));
683 prev_path = new_path;
684 prev_subgraph = new_subgraph;
685 }
686 }
687 digest.push(fetch_dependency_graph);
688 Ok(digest)
689}
690
691fn only_root_subgraph(graph: &FetchDependencyGraph) -> Result<Arc<str>, FederationError> {
692 let mut iter = graph.root_node_by_subgraph_iter();
693 let (Some((name, _)), None) = (iter.next(), iter.next()) else {
694 return Err(FederationError::internal(format!(
695 "{graph} should have only one root."
696 )));
697 };
698 Ok(name.clone())
699}
700
701#[cfg_attr(
702 feature = "snapshot_tracing",
703 tracing::instrument(level = "trace", skip_all, name = "compute_root_fetch_groups")
704)]
705pub(crate) fn compute_root_fetch_groups(
706 root_kind: SchemaRootDefinitionKind,
707 federated_query_graph: &QueryGraph,
708 dependency_graph: &mut FetchDependencyGraph,
709 path: &OpPathTree,
710 type_conditioned_fetching_enabled: bool,
711 check_cancellation: &dyn Fn() -> Result<(), SingleFederationError>,
712) -> Result<(), FederationError> {
713 for child in &path.childs {
720 let edge = child.edge.expect("The root edge should not be None");
721 let (_source_node, target_node) = path.graph.edge_endpoints(edge)?;
722 let target_node = path.graph.node_weight(target_node)?;
723 let subgraph_name = &target_node.source;
724 let root_type: CompositeTypeDefinitionPosition = match &target_node.type_ {
725 QueryGraphNodeType::SchemaType(OutputTypeDefinitionPosition::Object(object)) => {
726 object.clone().into()
727 }
728 ty => {
729 return Err(FederationError::internal(format!(
730 "expected an object type for the root of a subgraph, found {ty}"
731 )));
732 }
733 };
734 let fetch_dependency_node = dependency_graph.get_or_create_root_node(
735 subgraph_name,
736 root_kind,
737 root_type.clone(),
738 )?;
739 snapshot!(
740 "FetchDependencyGraph",
741 dependency_graph.to_dot(),
742 "tree_with_root_node"
743 );
744 let subgraph_schema = federated_query_graph.schema_by_source(subgraph_name)?;
745 let supergraph_root_type = convert_type_from_subgraph(
746 root_type,
747 subgraph_schema,
748 &dependency_graph.supergraph_schema,
749 )?;
750 compute_nodes_for_tree(
751 dependency_graph,
752 &child.tree,
753 fetch_dependency_node,
754 FetchDependencyGraphNodePath::new(
755 dependency_graph.supergraph_schema.clone(),
756 type_conditioned_fetching_enabled,
757 supergraph_root_type,
758 )?,
759 Default::default(),
760 &Default::default(),
761 check_cancellation,
762 )?;
763 }
764 Ok(())
765}
766
767fn compute_root_parallel_dependency_graph(
768 parameters: &QueryPlanningParameters,
769 has_defers: bool,
770 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
771 resolver_cache: &mut ConditionResolverCache,
772) -> Result<(FetchDependencyGraph, QueryPlanCost), FederationError> {
773 trace!("Starting process to construct a parallel fetch dependency graph");
774 let selection_set = parameters.operation.selection_set.clone();
775 let best_plan = compute_root_parallel_best_plan(
776 parameters,
777 selection_set,
778 has_defers,
779 non_local_selection_state,
780 resolver_cache,
781 )?;
782 snapshot!(
783 "FetchDependencyGraph",
784 best_plan.fetch_dependency_graph.to_dot(),
785 "Fetch dependency graph returned from compute_root_parallel_best_plan"
786 );
787 Ok((best_plan.fetch_dependency_graph, best_plan.cost))
788}
789
790fn compute_root_parallel_best_plan(
791 parameters: &QueryPlanningParameters,
792 selection: SelectionSet,
793 has_defers: bool,
794 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
795 resolver_cache: &mut ConditionResolverCache,
796) -> Result<BestQueryPlanInfo, FederationError> {
797 let planning_traversal = QueryPlanningTraversal::new(
798 parameters,
799 selection,
800 has_defers,
801 parameters.operation.root_kind,
802 FetchDependencyGraphToCostProcessor,
803 non_local_selection_state.as_mut(),
804 None,
805 resolver_cache,
806 )?;
807
808 Ok(planning_traversal
811 .find_best_plan()?
812 .unwrap_or_else(|| BestQueryPlanInfo::empty(parameters)))
813}
814
815fn compute_root_parallel_best_plan_for_mutation(
816 parameters: &QueryPlanningParameters,
817 selection: SelectionSet,
818 has_defers: bool,
819 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
820 resolver_cache: &mut ConditionResolverCache,
821) -> Result<BestQueryPlanInfo, FederationError> {
822 parameters.federated_query_graph.out_edges(parameters.head).into_iter().map(|edge_ref| {
823 let mutation_subgraph = parameters.federated_query_graph.node_weight(edge_ref.target())?.source.clone();
824 let planning_traversal = QueryPlanningTraversal::new(
825 parameters,
826 selection.clone(),
827 has_defers,
828 parameters.operation.root_kind,
829 FetchDependencyGraphToCostProcessor,
830 non_local_selection_state.as_mut(),
831 Some(mutation_subgraph),
832 resolver_cache,
833 )?;
834 planning_traversal.find_best_plan()
835 }).process_results(|iter| iter
836 .flatten()
837 .min_by(|a, b| a.cost.total_cmp(&b.cost))
838 .map(Ok)
839 .unwrap_or_else(|| {
840 if parameters.disabled_subgraphs.is_empty() {
841 Err(FederationError::internal(format!(
842 "Was not able to plan {} starting from a single subgraph: This shouldn't have happened.",
843 parameters.operation,
844 )))
845 } else {
846 Err(SingleFederationError::NoPlanFoundWithDisabledSubgraphs.into())
849 }
850 })
851 )?
852}
853
854fn compute_plan_internal(
855 parameters: &mut QueryPlanningParameters,
856 processor: &mut FetchDependencyGraphToQueryPlanProcessor,
857 has_defers: bool,
858 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
859 resolver_cache: &mut ConditionResolverCache,
860) -> Result<(Option<PlanNode>, QueryPlanCost), FederationError> {
861 let root_kind = parameters.operation.root_kind;
862
863 let (main, deferred, primary_selection, cost) = if root_kind
864 == SchemaRootDefinitionKind::Mutation
865 {
866 let dependency_graphs = compute_root_serial_dependency_graph_for_mutation(
867 parameters,
868 has_defers,
869 non_local_selection_state,
870 resolver_cache,
871 )?;
872 let mut main = None;
873 let mut deferred = vec![];
874 let mut primary_selection = None::<SelectionSet>;
875 for mut dependency_graph in dependency_graphs {
876 let (local_main, local_deferred) =
877 dependency_graph.process(&mut *processor, root_kind)?;
878 main = match main {
879 Some(unlocal_main) => processor.reduce_sequence([Some(unlocal_main), local_main]),
880 None => local_main,
881 };
882 deferred.extend(local_deferred);
883 let new_selection = dependency_graph.defer_tracking.primary_selection;
884 match primary_selection.as_mut() {
885 Some(selection) => {
886 if let Some(new_selection) = new_selection {
887 selection.add_local_selection_set(&new_selection)?
888 }
889 }
890 None => primary_selection = new_selection,
891 }
892 }
893 (main, deferred, primary_selection, f64::NAN)
895 } else {
896 let (mut dependency_graph, cost) = compute_root_parallel_dependency_graph(
897 parameters,
898 has_defers,
899 non_local_selection_state,
900 resolver_cache,
901 )?;
902
903 let (main, deferred) = dependency_graph.process(&mut *processor, root_kind)?;
904 snapshot!(
905 "FetchDependencyGraph",
906 dependency_graph.to_dot(),
907 "Plan after calling FetchDependencyGraph::process"
908 );
909 let primary_selection = dependency_graph.defer_tracking.primary_selection;
911
912 (main, deferred, primary_selection, cost)
913 };
914
915 if deferred.is_empty() {
916 Ok((main, cost))
917 } else {
918 let Some(primary_selection) = primary_selection else {
919 unreachable!("Should have had a primary selection created");
920 };
921 let reduced_main = processor.reduce_defer(main, &primary_selection, deferred)?;
922 Ok((reduced_main, cost))
923 }
924}
925
926fn compute_plan_for_defer_conditionals(
927 parameters: &mut QueryPlanningParameters,
928 processor: &mut FetchDependencyGraphToQueryPlanProcessor,
929 defer_conditions: IndexMap<Name, IndexSet<String>>,
930 non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
931 resolver_cache: &mut ConditionResolverCache,
932) -> Result<(Option<PlanNode>, QueryPlanCost), FederationError> {
933 generate_condition_nodes(
934 parameters.operation.clone(),
935 defer_conditions.iter(),
936 &mut |op| {
937 parameters.operation = op;
938 compute_plan_internal(
939 parameters,
940 processor,
941 true,
942 non_local_selection_state,
943 resolver_cache,
944 )
945 },
946 )
947}
948
949fn generate_condition_nodes<'a>(
950 op: Arc<Operation>,
951 mut conditions: impl Clone + Iterator<Item = (&'a Name, &'a IndexSet<String>)>,
952 on_final_operation: &mut impl FnMut(
953 Arc<Operation>,
954 ) -> Result<(Option<PlanNode>, f64), FederationError>,
955) -> Result<(Option<PlanNode>, f64), FederationError> {
956 match conditions.next() {
957 None => on_final_operation(op),
958 Some((cond, labels)) => {
959 let else_op = Arc::unwrap_or_clone(op.clone()).reduce_defer(labels)?;
960 let if_op = op;
961 let (if_node, if_cost) =
962 generate_condition_nodes(if_op, conditions.clone(), on_final_operation)?;
963 let (else_node, else_cost) = generate_condition_nodes(
964 Arc::new(else_op),
965 conditions.clone(),
966 on_final_operation,
967 )?;
968 let node = ConditionNode {
969 condition_variable: cond.clone(),
970 if_clause: if_node.map(Box::new),
971 else_clause: else_node.map(Box::new),
972 };
973 Ok((
974 Some(PlanNode::Condition(Box::new(node))),
975 if_cost.max(else_cost),
976 ))
977 }
978 }
979}
980
981pub(crate) enum SubgraphOperationCompression {
982 GenerateFragments,
983 Disabled,
984}
985
986impl SubgraphOperationCompression {
987 pub(crate) fn compress(
989 &mut self,
990 operation: Operation,
991 ) -> Result<Valid<ExecutableDocument>, FederationError> {
992 match self {
993 Self::GenerateFragments => Ok(operation.generate_fragments()?),
994 Self::Disabled => {
995 let operation_document = operation.try_into().map_err(|err: FederationError| {
996 if err.has_invalid_graphql_error() {
997 internal_error!(
998 "Query planning produced an invalid subgraph operation.\n{err}"
999 )
1000 } else {
1001 err
1002 }
1003 })?;
1004 Ok(operation_document)
1005 }
1006 }
1007 }
1008}
1009
1010#[cfg(test)]
1011mod tests {
1012 use super::*;
1013
1014 const TEST_SUPERGRAPH: &str = r#"
1015schema
1016 @link(url: "https://specs.apollo.dev/link/v1.0")
1017 @link(url: "https://specs.apollo.dev/join/v0.2", for: EXECUTION)
1018{
1019 query: Query
1020}
1021
1022directive @join__field(graph: join__Graph!, requires: join__FieldSet, provides: join__FieldSet, type: String, external: Boolean, override: String, usedOverridden: Boolean) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
1023
1024directive @join__graph(name: String!, url: String!) on ENUM_VALUE
1025
1026directive @join__implements(graph: join__Graph!, interface: String!) repeatable on OBJECT | INTERFACE
1027
1028directive @join__type(graph: join__Graph!, key: join__FieldSet, extension: Boolean! = false, resolvable: Boolean! = true) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
1029
1030directive @link(url: String, as: String, for: link__Purpose, import: [link__Import]) repeatable on SCHEMA
1031
1032type Book implements Product
1033 @join__implements(graph: PRODUCTS, interface: "Product")
1034 @join__implements(graph: REVIEWS, interface: "Product")
1035 @join__type(graph: PRODUCTS, key: "id")
1036 @join__type(graph: REVIEWS, key: "id")
1037{
1038 id: ID!
1039 price: Price @join__field(graph: PRODUCTS)
1040 title: String @join__field(graph: PRODUCTS)
1041 vendor: User @join__field(graph: PRODUCTS)
1042 pages: Int @join__field(graph: PRODUCTS)
1043 avg_rating: Int @join__field(graph: PRODUCTS, requires: "reviews { rating }")
1044 reviews: [Review] @join__field(graph: PRODUCTS, external: true) @join__field(graph: REVIEWS)
1045}
1046
1047enum Currency
1048 @join__type(graph: PRODUCTS)
1049{
1050 USD
1051 EUR
1052}
1053
1054scalar join__FieldSet
1055
1056enum join__Graph {
1057 ACCOUNTS @join__graph(name: "accounts", url: "")
1058 PRODUCTS @join__graph(name: "products", url: "")
1059 REVIEWS @join__graph(name: "reviews", url: "")
1060}
1061
1062scalar link__Import
1063
1064enum link__Purpose {
1065 """
1066 `SECURITY` features provide metadata necessary to securely resolve fields.
1067 """
1068 SECURITY
1069
1070 """
1071 `EXECUTION` features provide metadata necessary for operation execution.
1072 """
1073 EXECUTION
1074}
1075
1076type Movie implements Product
1077 @join__implements(graph: PRODUCTS, interface: "Product")
1078 @join__implements(graph: REVIEWS, interface: "Product")
1079 @join__type(graph: PRODUCTS, key: "id")
1080 @join__type(graph: REVIEWS, key: "id")
1081{
1082 id: ID!
1083 price: Price @join__field(graph: PRODUCTS)
1084 title: String @join__field(graph: PRODUCTS)
1085 vendor: User @join__field(graph: PRODUCTS)
1086 length_minutes: Int @join__field(graph: PRODUCTS)
1087 avg_rating: Int @join__field(graph: PRODUCTS, requires: "reviews { rating }")
1088 reviews: [Review] @join__field(graph: PRODUCTS, external: true) @join__field(graph: REVIEWS)
1089}
1090
1091type Price
1092 @join__type(graph: PRODUCTS)
1093{
1094 value: Int
1095 currency: Currency
1096}
1097
1098interface Product
1099 @join__type(graph: PRODUCTS)
1100 @join__type(graph: REVIEWS)
1101{
1102 id: ID!
1103 price: Price @join__field(graph: PRODUCTS)
1104 vendor: User @join__field(graph: PRODUCTS)
1105 avg_rating: Int @join__field(graph: PRODUCTS)
1106 reviews: [Review] @join__field(graph: REVIEWS)
1107}
1108
1109type Query
1110 @join__type(graph: ACCOUNTS)
1111 @join__type(graph: PRODUCTS)
1112 @join__type(graph: REVIEWS)
1113{
1114 userById(id: ID!): User @join__field(graph: ACCOUNTS)
1115 me: User! @join__field(graph: ACCOUNTS) @join__field(graph: REVIEWS)
1116 productById(id: ID!): Product @join__field(graph: PRODUCTS)
1117 search(filter: SearchFilter): [Product] @join__field(graph: PRODUCTS)
1118 bestRatedProducts(limit: Int): [Product] @join__field(graph: REVIEWS)
1119}
1120
1121type Review
1122 @join__type(graph: PRODUCTS)
1123 @join__type(graph: REVIEWS)
1124{
1125 rating: Int @join__field(graph: PRODUCTS, external: true) @join__field(graph: REVIEWS)
1126 product: Product @join__field(graph: REVIEWS)
1127 author: User @join__field(graph: REVIEWS)
1128 text: String @join__field(graph: REVIEWS)
1129}
1130
1131input SearchFilter
1132 @join__type(graph: PRODUCTS)
1133{
1134 pattern: String!
1135 vendorName: String
1136}
1137
1138type User
1139 @join__type(graph: ACCOUNTS, key: "id")
1140 @join__type(graph: PRODUCTS, key: "id", resolvable: false)
1141 @join__type(graph: REVIEWS, key: "id")
1142{
1143 id: ID!
1144 name: String @join__field(graph: ACCOUNTS)
1145 email: String @join__field(graph: ACCOUNTS)
1146 password: String @join__field(graph: ACCOUNTS)
1147 nickname: String @join__field(graph: ACCOUNTS, override: "reviews")
1148 reviews: [Review] @join__field(graph: REVIEWS)
1149}
1150 "#;
1151
1152 #[test]
1153 fn plan_simple_query_for_single_subgraph() {
1154 let supergraph = Supergraph::new(TEST_SUPERGRAPH).unwrap();
1155 let planner = QueryPlanner::new(&supergraph, Default::default()).unwrap();
1156
1157 let document = ExecutableDocument::parse_and_validate(
1158 planner.api_schema().schema(),
1159 r#"
1160 {
1161 userById(id: 1) {
1162 name
1163 email
1164 }
1165 }
1166 "#,
1167 "operation.graphql",
1168 )
1169 .unwrap();
1170 let plan = planner
1171 .build_query_plan(&document, None, Default::default())
1172 .unwrap();
1173 insta::assert_snapshot!(plan, @r###"
1174 QueryPlan {
1175 Fetch(service: "accounts") {
1176 {
1177 userById(id: 1) {
1178 name
1179 email
1180 }
1181 }
1182 },
1183 }
1184 "###);
1185 }
1186
1187 #[test]
1188 fn plan_simple_query_for_multiple_subgraphs() {
1189 let supergraph = Supergraph::new(TEST_SUPERGRAPH).unwrap();
1190 let planner = QueryPlanner::new(&supergraph, Default::default()).unwrap();
1191
1192 let document = ExecutableDocument::parse_and_validate(
1193 planner.api_schema().schema(),
1194 r#"
1195 {
1196 bestRatedProducts {
1197 vendor { name }
1198 }
1199 }
1200 "#,
1201 "operation.graphql",
1202 )
1203 .unwrap();
1204 let plan = planner
1205 .build_query_plan(&document, None, Default::default())
1206 .unwrap();
1207 insta::assert_snapshot!(plan, @r###"
1208 QueryPlan {
1209 Sequence {
1210 Fetch(service: "reviews") {
1211 {
1212 bestRatedProducts {
1213 __typename
1214 ... on Book {
1215 __typename
1216 id
1217 }
1218 ... on Movie {
1219 __typename
1220 id
1221 }
1222 }
1223 }
1224 },
1225 Flatten(path: "bestRatedProducts.@") {
1226 Fetch(service: "products") {
1227 {
1228 ... on Book {
1229 __typename
1230 id
1231 }
1232 ... on Movie {
1233 __typename
1234 id
1235 }
1236 } =>
1237 {
1238 ... on Book {
1239 vendor {
1240 __typename
1241 id
1242 }
1243 }
1244 ... on Movie {
1245 vendor {
1246 __typename
1247 id
1248 }
1249 }
1250 }
1251 },
1252 },
1253 Flatten(path: "bestRatedProducts.@.vendor") {
1254 Fetch(service: "accounts") {
1255 {
1256 ... on User {
1257 __typename
1258 id
1259 }
1260 } =>
1261 {
1262 ... on User {
1263 name
1264 }
1265 }
1266 },
1267 },
1268 },
1269 }
1270 "###);
1271 }
1272
1273 #[test]
1274 fn plan_simple_root_field_query_for_multiple_subgraphs() {
1275 let supergraph = Supergraph::new(TEST_SUPERGRAPH).unwrap();
1276 let planner = QueryPlanner::new(&supergraph, Default::default()).unwrap();
1277
1278 let document = ExecutableDocument::parse_and_validate(
1279 planner.api_schema().schema(),
1280 r#"
1281 {
1282 userById(id: 1) {
1283 name
1284 email
1285 }
1286 bestRatedProducts {
1287 id
1288 avg_rating
1289 }
1290 }
1291 "#,
1292 "operation.graphql",
1293 )
1294 .unwrap();
1295 let plan = planner
1296 .build_query_plan(&document, None, Default::default())
1297 .unwrap();
1298 insta::assert_snapshot!(plan, @r###"
1299 QueryPlan {
1300 Parallel {
1301 Fetch(service: "accounts") {
1302 {
1303 userById(id: 1) {
1304 name
1305 email
1306 }
1307 }
1308 },
1309 Sequence {
1310 Fetch(service: "reviews") {
1311 {
1312 bestRatedProducts {
1313 __typename
1314 id
1315 ... on Book {
1316 __typename
1317 id
1318 reviews {
1319 rating
1320 }
1321 }
1322 ... on Movie {
1323 __typename
1324 id
1325 reviews {
1326 rating
1327 }
1328 }
1329 }
1330 }
1331 },
1332 Flatten(path: "bestRatedProducts.@") {
1333 Fetch(service: "products") {
1334 {
1335 ... on Book {
1336 __typename
1337 id
1338 reviews {
1339 rating
1340 }
1341 }
1342 ... on Movie {
1343 __typename
1344 id
1345 reviews {
1346 rating
1347 }
1348 }
1349 } =>
1350 {
1351 ... on Book {
1352 avg_rating
1353 }
1354 ... on Movie {
1355 avg_rating
1356 }
1357 }
1358 },
1359 },
1360 },
1361 },
1362 }
1363 "###);
1364 }
1365
1366 #[test]
1367 fn test_optimize_no_fragments_generated() {
1368 let supergraph = Supergraph::new(TEST_SUPERGRAPH).unwrap();
1369 let api_schema = supergraph.to_api_schema(Default::default()).unwrap();
1370 let document = ExecutableDocument::parse_and_validate(
1371 api_schema.schema(),
1372 r#"
1373 {
1374 userById(id: 1) {
1375 id
1376 ...userFields
1377 },
1378 another_user: userById(id: 2) {
1379 name
1380 email
1381 }
1382 }
1383 fragment userFields on User {
1384 name
1385 email
1386 }
1387 "#,
1388 "operation.graphql",
1389 )
1390 .unwrap();
1391
1392 let config = QueryPlannerConfig {
1393 generate_query_fragments: true,
1394 ..Default::default()
1395 };
1396 let planner = QueryPlanner::new(&supergraph, config).unwrap();
1397 let plan = planner
1398 .build_query_plan(&document, None, Default::default())
1399 .unwrap();
1400 insta::assert_snapshot!(plan, @r###"
1401 QueryPlan {
1402 Fetch(service: "accounts") {
1403 {
1404 userById(id: 1) {
1405 id
1406 name
1407 email
1408 }
1409 another_user: userById(id: 2) {
1410 name
1411 email
1412 }
1413 }
1414 },
1415 }
1416 "###);
1417 }
1418
1419 #[test]
1420 fn drop_operation_root_level_typename() {
1421 let supergraph = Supergraph::new(TEST_SUPERGRAPH).unwrap();
1422 let planner = QueryPlanner::new(&supergraph, Default::default()).unwrap();
1423
1424 let document = ExecutableDocument::parse_and_validate(
1425 planner.api_schema().schema(),
1426 r#"
1427 {
1428 __typename
1429 bestRatedProducts {
1430 id
1431 }
1432 }
1433 "#,
1434 "operation.graphql",
1435 )
1436 .unwrap();
1437 let plan = planner
1438 .build_query_plan(&document, None, Default::default())
1439 .unwrap();
1440 insta::assert_snapshot!(plan, @r###"
1442 QueryPlan {
1443 Fetch(service: "reviews") {
1444 {
1445 bestRatedProducts {
1446 __typename
1447 id
1448 }
1449 }
1450 },
1451 }
1452 "###);
1453 }
1454
1455 #[test]
1456 fn test_query_plan_statistics_nan_cost() {
1457 let stats = QueryPlanningStatistics {
1458 evaluated_plan_count: Cell::new(10),
1459 evaluated_plan_paths: Cell::new(20),
1460 best_plan_cost: f64::NAN,
1461 };
1462 let serialized = serde_json::to_string_pretty(&stats).expect("Serializing");
1463 insta::assert_snapshot!(serialized, @r###"
1464 {
1465 "evaluated_plan_count": 10,
1466 "evaluated_plan_paths": 20,
1467 "best_plan_cost": null
1468 }
1469 "###);
1470
1471 let deserialized: QueryPlanningStatistics =
1472 serde_json::from_str(&serialized).expect("Deserializing");
1473 assert!(deserialized.best_plan_cost.is_nan());
1474 }
1475}