CVE-2025-32033

@@ -32,6 +32,7 @@ use crate::query_graph::QueryGraphEdge;
 use crate::query_graph::QueryGraphEdgeTransition;
 use crate::query_graph::QueryGraphNode;
 use crate::query_graph::QueryGraphNodeType;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation::precompute_non_local_selection_metadata;
 use crate::schema::ValidFederationSchema;
 use crate::schema::field_set::parse_field_set;
 use crate::schema::position::AbstractTypeDefinitionPosition;
@@ -77,6 +78,7 @@ pub fn build_federated_query_graph(
         root_kinds_to_nodes_by_source: Default::default(),
         non_trivial_followup_edges: Default::default(),
         arguments_to_context_ids_by_source: Default::default(),
+        non_local_selection_metadata: Default::default(),
     };
     let query_graph =
         extract_subgraphs_from_supergraph(&supergraph_schema, validate_extracted_subgraphs)?
@@ -112,6 +114,7 @@ pub fn build_query_graph(
         root_kinds_to_nodes_by_source: Default::default(),
         non_trivial_followup_edges: Default::default(),
         arguments_to_context_ids_by_source: Default::default(),
+        non_local_selection_metadata: Default::default(),
     };
     let builder = SchemaQueryGraphBuilder::new(query_graph, name, schema, None, false)?;
     query_graph = builder.build()?;
@@ -1002,6 +1005,10 @@ impl FederatedQueryGraphBuilder {
         self.handle_interface_object()?;
         // This method adds no nodes/edges, but just precomputes followup edge information.
         self.precompute_non_trivial_followup_edges()?;
+        // This method adds no nodes/edges, but just precomputes metadata for estimating the count
+        // of non_local_selections.
+        self.base.query_graph.non_local_selection_metadata =
+            precompute_non_local_selection_metadata(&self.base.query_graph)?;
         Ok(self.base.build())
     }
 

@@ -484,6 +484,15 @@ impl OpPathElement {
         }
     }
 
+    pub(crate) fn has_defer(&self) -> bool {
+        match self {
+            OpPathElement::Field(_) => false,
+            OpPathElement::InlineFragment(inline_fragment) => {
+                inline_fragment.directives.has("defer")
+            }
+        }
+    }
+
     /// Returns this fragment element but with any @defer directive on it removed.
     ///
     /// This method will return `None` if, upon removing @defer, the fragment has no conditions nor

@@ -50,6 +50,7 @@ use crate::query_graph::graph_path::OpGraphPathTrigger;
 use crate::query_graph::graph_path::OpPathElement;
 use crate::query_plan::QueryPlanCost;
 use crate::query_plan::query_planner::EnabledOverrideConditions;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub(crate) struct QueryGraphNode {
@@ -201,7 +202,7 @@ impl QueryGraphEdge {
         conditions_to_check: &EnabledOverrideConditions,
     ) -> bool {
         if let Some(override_condition) = &self.override_condition {
-            override_condition.condition == conditions_to_check.contains(&override_condition.label)
+            override_condition.check(conditions_to_check)
         } else {
             true
         }
@@ -238,6 +239,12 @@ pub(crate) struct OverrideCondition {
     pub(crate) condition: bool,
 }
 
+impl OverrideCondition {
+    pub(crate) fn check(&self, enabled_conditions: &EnabledOverrideConditions) -> bool {
+        self.condition == enabled_conditions.contains(&self.label)
+    }
+}
+
 impl Display for OverrideCondition {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(f, "{} = {}", self.label, self.condition)
@@ -405,6 +412,9 @@ pub struct QueryGraph {
     /// argument coordinates). This identifier is called the "context ID".
     arguments_to_context_ids_by_source:
         IndexMap<Arc<str>, IndexMap<ObjectFieldArgumentDefinitionPosition, Name>>,
+    /// To speed up the estimation of counting non-local selections, we precompute specific metadata
+    /// about the query graph and store that here.
+    non_local_selection_metadata: non_local_selections_estimation::QueryGraphMetadata,
 }
 
 impl QueryGraph {
@@ -500,7 +510,7 @@ impl QueryGraph {
         self.types_to_nodes_by_source(&self.current_source)
     }
 
-    fn types_to_nodes_by_source(
+    pub(super) fn types_to_nodes_by_source(
         &self,
         source: &str,
     ) -> Result<&IndexMap<NamedType, IndexSet<NodeIndex>>, FederationError> {
@@ -582,6 +592,12 @@ impl QueryGraph {
         !self.arguments_to_context_ids_by_source.is_empty()
     }
 
+    pub(crate) fn non_local_selection_metadata(
+        &self,
+    ) -> &non_local_selections_estimation::QueryGraphMetadata {
+        &self.non_local_selection_metadata
+    }
+
     /// All outward edges from the given node (including self-key and self-root-type-resolution
     /// edges). Primarily used by `@defer`, when needing to re-enter a subgraph for a deferred
     /// section.

@@ -44,6 +44,7 @@ use crate::query_plan::query_planning_traversal::BestQueryPlanInfo;
 use crate::query_plan::query_planning_traversal::QueryPlanningParameters;
 use crate::query_plan::query_planning_traversal::QueryPlanningTraversal;
 use crate::query_plan::query_planning_traversal::convert_type_from_subgraph;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation;
 use crate::schema::ValidFederationSchema;
 use crate::schema::position::AbstractTypeDefinitionPosition;
 use crate::schema::position::CompositeTypeDefinitionPosition;
@@ -172,7 +173,7 @@ pub struct QueryPlanningStatistics {
     pub evaluated_plan_paths: Cell<usize>,
 }
 
-#[derive(Default, Clone)]
+#[derive(Clone)]
 pub struct QueryPlanOptions<'a> {
     /// A set of labels which will be used _during query planning_ to
     /// enable/disable edges with a matching label in their override condition.
@@ -183,6 +184,19 @@ pub struct QueryPlanOptions<'a> {
     pub override_conditions: Vec<String>,
 
     pub check_for_cooperative_cancellation: Option<&'a dyn Fn() -> ControlFlow<()>>,
+    /// Impose a limit on the number of non-local selections, which can be a
+    /// performance hazard. On by default.
+    pub non_local_selections_limit_enabled: bool,
+}
+
+impl Default for QueryPlanOptions<'_> {
+    fn default() -> Self {
+        Self {
+            override_conditions: Vec::new(),
+            check_for_cooperative_cancellation: None,
+            non_local_selections_limit_enabled: true,
+        }
+    }
 }
 
 impl std::fmt::Debug for QueryPlanOptions<'_> {
@@ -197,6 +211,10 @@ impl std::fmt::Debug for QueryPlanOptions<'_> {
                     &"None"
                 },
             )
+            .field(
+                "non_local_selections_limit_enabled",
+                &self.non_local_selections_limit_enabled,
+            )
             .finish()
     }
 }
@@ -224,7 +242,7 @@ pub struct QueryPlanner {
     /// subgraphs.
     // PORT_NOTE: Named `inconsistentAbstractTypesRuntimes` in the JS codebase, which was slightly
     // confusing.
-    abstract_types_with_inconsistent_runtime_types: IndexSet<AbstractTypeDefinitionPosition>,
+    abstract_types_with_inconsistent_runtime_types: IndexSet<Name>,
 }
 
 impl QueryPlanner {
@@ -317,6 +335,7 @@ impl QueryPlanner {
             .get_types()
             .filter_map(|position| AbstractTypeDefinitionPosition::try_from(position).ok())
             .filter(|position| is_inconsistent(position.clone()))
+            .map(|position| position.type_name().clone())
             .collect::<IndexSet<_>>();
 
         Ok(Self {
@@ -443,10 +462,23 @@ impl QueryPlanner {
             fetch_id_generator: Arc::new(FetchIdGenerator::new()),
         };
 
+        let mut non_local_selection_state = options
+            .non_local_selections_limit_enabled
+            .then(non_local_selections_estimation::State::default);
         let root_node = if !defer_conditions.is_empty() {
-            compute_plan_for_defer_conditionals(&mut parameters, &mut processor, defer_conditions)
+            compute_plan_for_defer_conditionals(
+                &mut parameters,
+                &mut processor,
+                defer_conditions,
+                &mut non_local_selection_state,
+            )
         } else {
-            compute_plan_internal(&mut parameters, &mut processor, has_defers)
+            compute_plan_internal(
+                &mut parameters,
+                &mut processor,
+                has_defers,
+                &mut non_local_selection_state,
+            )
         }?;
 
         let root_node = match root_node {
@@ -521,6 +553,7 @@ impl QueryPlanner {
 fn compute_root_serial_dependency_graph(
     parameters: &QueryPlanningParameters,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Vec<FetchDependencyGraph>, FederationError> {
     let QueryPlanningParameters {
         supergraph_schema,
@@ -547,14 +580,24 @@ fn compute_root_serial_dependency_graph(
         mut fetch_dependency_graph,
         path_tree: mut prev_path,
         ..
-    } = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+    } = compute_root_parallel_best_plan(
+        parameters,
+        selection_set,
+        has_defers,
+        non_local_selection_state,
+    )?;
     let mut prev_subgraph = only_root_subgraph(&fetch_dependency_graph)?;
     for selection_set in split_roots {
         let BestQueryPlanInfo {
             fetch_dependency_graph: new_dep_graph,
             path_tree: new_path,
             ..
-        } = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+        } = compute_root_parallel_best_plan(
+            parameters,
+            selection_set,
+            has_defers,
+            non_local_selection_state,
+        )?;
         let new_subgraph = only_root_subgraph(&new_dep_graph)?;
         if new_subgraph == prev_subgraph {
             // The new operation (think 'mutation' operation) is on the same subgraph than the previous one, so we can concat them in a single fetch
@@ -677,10 +720,16 @@ pub(crate) fn compute_root_fetch_groups(
 fn compute_root_parallel_dependency_graph(
     parameters: &QueryPlanningParameters,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<FetchDependencyGraph, FederationError> {
     trace!("Starting process to construct a parallel fetch dependency graph");
     let selection_set = parameters.operation.selection_set.clone();
-    let best_plan = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+    let best_plan = compute_root_parallel_best_plan(
+        parameters,
+        selection_set,
+        has_defers,
+        non_local_selection_state,
+    )?;
     snapshot!(
         "FetchDependencyGraph",
         best_plan.fetch_dependency_graph.to_dot(),
@@ -693,13 +742,15 @@ fn compute_root_parallel_best_plan(
     parameters: &QueryPlanningParameters,
     selection: SelectionSet,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<BestQueryPlanInfo, FederationError> {
     let planning_traversal = QueryPlanningTraversal::new(
         parameters,
         selection,
         has_defers,
         parameters.operation.root_kind,
         FetchDependencyGraphToCostProcessor,
+        non_local_selection_state.as_mut(),
     )?;
 
     // Getting no plan means the query is essentially unsatisfiable (it's a valid query, but we can prove it will never return a result),
@@ -713,11 +764,16 @@ fn compute_plan_internal(
     parameters: &mut QueryPlanningParameters,
     processor: &mut FetchDependencyGraphToQueryPlanProcessor,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Option<PlanNode>, FederationError> {
     let root_kind = parameters.operation.root_kind;
 
     let (main, deferred, primary_selection) = if root_kind == SchemaRootDefinitionKind::Mutation {
-        let dependency_graphs = compute_root_serial_dependency_graph(parameters, has_defers)?;
+        let dependency_graphs = compute_root_serial_dependency_graph(
+            parameters,
+            has_defers,
+            non_local_selection_state,
+        )?;
         let mut main = None;
         let mut deferred = vec![];
         let mut primary_selection = None::<SelectionSet>;
@@ -741,7 +797,11 @@ fn compute_plan_internal(
         }
         (main, deferred, primary_selection)
     } else {
-        let mut dependency_graph = compute_root_parallel_dependency_graph(parameters, has_defers)?;
+        let mut dependency_graph = compute_root_parallel_dependency_graph(
+            parameters,
+            has_defers,
+            non_local_selection_state,
+        )?;
 
         let (main, deferred) = dependency_graph.process(&mut *processor, root_kind)?;
         snapshot!(
@@ -769,13 +829,14 @@ fn compute_plan_for_defer_conditionals(
     parameters: &mut QueryPlanningParameters,
     processor: &mut FetchDependencyGraphToQueryPlanProcessor,
     defer_conditions: IndexMap<Name, IndexSet<String>>,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Option<PlanNode>, FederationError> {
     generate_condition_nodes(
         parameters.operation.clone(),
         defer_conditions.iter(),
         &mut |op| {
             parameters.operation = op;
-            compute_plan_internal(parameters, processor, true)
+            compute_plan_internal(parameters, processor, true, non_local_selection_state)
         },
     )
 }

@@ -0,0 +1,991 @@
+use apollo_compiler::Name;
+use apollo_compiler::collections::IndexMap;
+use apollo_compiler::collections::IndexSet;
+use apollo_compiler::schema::ExtendedType;
+use indexmap::map::Entry;
+use petgraph::graph::NodeIndex;
+use petgraph::visit::EdgeRef;
+use petgraph::visit::IntoNodeReferences;
+
+use crate::bail;
+use crate::error::FederationError;
+use crate::operation::Selection;
+use crate::operation::SelectionSet;
+use crate::query_graph::OverrideCondition;
+use crate::query_graph::QueryGraph;
+use crate::query_graph::QueryGraphEdgeTransition;
+use crate::query_graph::graph_path::OpPathElement;
+use crate::query_plan::query_planning_traversal::QueryPlanningTraversal;
+use crate::schema::position::CompositeTypeDefinitionPosition;
+use crate::schema::position::INTROSPECTION_TYPENAME_FIELD_NAME;
+use crate::schema::position::ObjectTypeDefinitionPosition;
+
+impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
+    pub(super) const MAX_NON_LOCAL_SELECTIONS: u64 = 100_000;
+
+    /// This calls `check_non_local_selections_limit_exceeded()` for each of the selections in the
+    /// open branches stack; see that function's doc comment for more information.
+    pub(super) fn check_non_local_selections_limit_exceeded_at_root(
+        &self,
+        state: &mut State,
+    ) -> Result<bool, FederationError> {
+        for branch in &self.open_branches {
+            let tail_nodes = branch
+                .open_branch
+                .0
+                .iter()
+                .flat_map(|option| option.paths.0.iter().map(|path| path.tail))
+                .collect::<IndexSet<_>>();
+            let tail_nodes_info = self.estimate_nodes_with_indirect_options(tail_nodes)?;
+
+            // Note that top-level selections aren't avoided via fully-local selection set
+            // optimization, so we always add them here.
+            if Self::update_count(
+                branch.selections.len(),
+                tail_nodes_info.next_nodes.len(),
+                state,
+            ) {
+                return Ok(true);
+            }
+
+            for selection in &branch.selections {
+                if let Some(selection_set) = selection.selection_set() {
+                    let selection_has_defer = selection.element().has_defer();
+                    let next_nodes = self.estimate_next_nodes_for_selection(
+                        &selection.element(),
+                        &tail_nodes_info,
+                        state,
+                    )?;
+                    if self.check_non_local_selections_limit_exceeded(
+                        selection_set,
+                        &next_nodes,
+                        selection_has_defer,
+                        state,
+                    )? {
+                        return Ok(true);
+                    }
+                }
+            }
+        }
+        Ok(false)
+    }
+
+    /// When recursing through a selection set to generate options from each element, there is an
+    /// optimization that allows us to avoid option exploration if a selection set is "fully local"
+    /// from all the possible nodes we could be at in the query graph.
+    ///
+    /// This function computes an approximate upper bound on the number of selections in a selection
+    /// set that wouldn't be avoided by such an optimization (i.e. the "non-local" selections), and
+    /// adds it to the given count in the state. Note that the count for a given selection set is
+    /// scaled by an approximate upper bound on the possible number of tail nodes for paths ending
+    /// at that selection set. If at any point, the count exceeds `Self::MAX_NON_LOCAL_SELECTIONS`,
+    /// then this function will return `true`.
+    ///
+    /// This function's code is closely related to `selection_set_is_fully_local_from_all_nodes()`
+    /// (which implements the aforementioned optimization). However, when it comes to traversing the
+    /// query graph, we generally ignore the effects of edge pre-conditions and other optimizations
+    /// to option generation for efficiency's sake, giving us an upper bound since the extra nodes
+    /// may fail some of the checks (e.g. the selection set may not rebase on them).
+    ///
+    /// Note that this function takes in whether the parent selection of the selection set has
+    /// @defer, as that affects whether the optimization is disabled for that selection set.
+    fn check_non_local_selections_limit_exceeded(
+        &self,
+        selection_set: &SelectionSet,
+        parent_nodes: &NextNodesInfo,
+        parent_selection_has_defer: bool,
+        state: &mut State,
+    ) -> Result<bool, FederationError> {
+        // Compute whether the selection set is non-local, and if so, add its selections to the
+        // count. Any of the following causes the selection set to be non-local.
+        // 1. The selection set's nodes having at least one reachable cross-subgraph edge.
+        // 2. The parent selection having @defer.
+        // 3. Any selection in the selection set having @defer.
+        // 4. Any selection in the selection set being an inline fragment whose type condition
+        //    has inconsistent runtime types across subgraphs.
+        // 5. Any selection in the selection set being unable to be rebased on the selection
+        //    set's nodes.
+        // 6. Any nested selection sets causing the count to be incremented.
+        let mut selection_set_is_non_local = parent_nodes
+            .next_nodes_have_reachable_cross_subgraph_edges
+            || parent_selection_has_defer;
+        for selection in selection_set.selections.values() {
+            let element = selection.element();
+            let selection_has_defer = element.has_defer();
+            let selection_has_inconsistent_runtime_types =
+                if let OpPathElement::InlineFragment(inline_fragment) = element {
+                    inline_fragment
+                        .type_condition_position
+                        .map(|type_condition_pos| {
+                            self.parameters
+                                .abstract_types_with_inconsistent_runtime_types
+                                .contains(type_condition_pos.type_name())
+                        })
+                        .unwrap_or_default()
+                } else {
+                    false
+                };
+
+            let old_count = state.count;
+            if let Some(selection_set) = selection.selection_set() {
+                let next_nodes = self.estimate_next_nodes_for_selection(
+                    &selection.element(),
+                    parent_nodes,
+                    state,
+                )?;
+                if self.check_non_local_selections_limit_exceeded(
+                    selection_set,
+                    &next_nodes,
+                    selection_has_defer,
+                    state,
+                )? {
+                    return Ok(true);
+                }
+            }
+
+            selection_set_is_non_local = selection_set_is_non_local
+                || selection_has_defer
+                || selection_has_inconsistent_runtime_types
+                || (old_count != state.count);
+        }
+        // Determine whether the selection can be rebased on all selection set nodes (without
+        // indirect options). This is more expensive, so we do this last/only if needed. Note
+        // that we were originally calling a slightly modified `can_add_to()` to mimic the logic
+        // in `selection_set_is_fully_local_from_all_nodes()`, but this ended up being rather
+        // expensive in practice, so an optimized version using precomputation is used below.
+        if !selection_set_is_non_local && !parent_nodes.next_nodes.is_empty() {
+            let metadata = self
+                .parameters
+                .federated_query_graph
+                .non_local_selection_metadata();
+            for selection in selection_set.selections.values() {
+                match selection {
+                    Selection::Field(field) => {
+                        // Note that while the precomputed metadata accounts for @fromContext,
+                        // it doesn't account for checking whether the operation field's parent
+                        // type either matches the subgraph schema's parent type name or is an
+                        // interface type. Given current composition rules, this should always
+                        // be the case when rebasing supergraph/API schema queries onto one of
+                        // its subgraph schema, so we avoid the check here for performance.
+                        let Some(rebaseable_parent_nodes) = metadata
+                            .fields_to_rebaseable_parent_nodes
+                            .get(field.field.name())
+                        else {
+                            selection_set_is_non_local = true;
+                            break;
+                        };
+                        if !parent_nodes.next_nodes.is_subset(rebaseable_parent_nodes) {
+                            selection_set_is_non_local = true;
+                            break;
+                        }
+                    }
+                    Selection::InlineFragment(inline_fragment) => {
+                        let Some(type_condition_pos) =
+                            &inline_fragment.inline_fragment.type_condition_position
+                        else {
+                            // Inline fragments without type conditions can always be rebased.
+                            continue;
+                        };
+                        let Some(rebaseable_parent_nodes) = metadata
+                            .inline_fragments_to_rebaseable_parent_nodes
+                            .get(type_condition_pos.type_name())
+                        else {
+                            selection_set_is_non_local = true;
+                            break;
+                        };
+                        if !parent_nodes.next_nodes.is_subset(rebaseable_parent_nodes) {
+                            selection_set_is_non_local = true;
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        if selection_set_is_non_local
+            && Self::update_count(
+                selection_set.selections.len(),
+                parent_nodes.next_nodes.len(),
+                state,
+            )
+        {
+            return Ok(true);
+        }
+        Ok(false)
+    }
+
+    /// Updates the non-local selection set count in the state, returning true if this causes the
+    /// count to exceed `Self::MAX_NON_LOCAL_SELECTIONS`.
+    fn update_count(num_selections: usize, num_parent_nodes: usize, state: &mut State) -> bool {
+        let Ok(num_selections) = u64::try_from(num_selections) else {
+            return true;
+        };
+        let Ok(num_parent_nodes) = u64::try_from(num_parent_nodes) else {
+            return true;
+        };
+        let Some(additional_count) = num_selections.checked_mul(num_parent_nodes) else {
+            return true;
+        };
+        if let Some(new_count) = state
+            .count
+            .checked_add(additional_count)
+            .take_if(|v| *v <= Self::MAX_NON_LOCAL_SELECTIONS)
+        {
+            state.count = new_count;
+        } else {
+            return true;
+        };
+        false
+    }
+
+    /// In `check_non_local_selections_limit_exceeded()`, when handling a given selection for a set
+    /// of parent nodes (including indirect options), this function can be used to estimate an
+    /// upper bound on the next nodes after taking the selection (also with indirect options).
+    fn estimate_next_nodes_for_selection(
+        &self,
+        element: &OpPathElement,
+        parent_nodes: &NextNodesInfo,
+        state: &mut State,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let cache = state
+            .next_nodes_cache
+            .entry(match element {
+                OpPathElement::Field(field) => {
+                    SelectionKey::Field(field.field_position.field_name().clone())
+                }
+                OpPathElement::InlineFragment(inline_fragment) => {
+                    let Some(type_condition_pos) = &inline_fragment.type_condition_position else {
+                        return Ok(parent_nodes.clone());
+                    };
+                    SelectionKey::InlineFragment(type_condition_pos.type_name().clone())
+                }
+            })
+            .or_default();
+        let mut next_nodes = NextNodesInfo::default();
+        for type_name in &parent_nodes.next_nodes_with_indirect_options.types {
+            match cache.types_to_next_nodes.entry(type_name.clone()) {
+                Entry::Occupied(entry) => next_nodes.extend(entry.get()),
+                Entry::Vacant(entry) => {
+                    let Some(indirect_options) = self
+                        .parameters
+                        .federated_query_graph
+                        .non_local_selection_metadata()
+                        .types_to_indirect_options
+                        .get(type_name)
+                    else {
+                        bail!("Unexpectedly missing node information for cached type.");
+                    };
+                    let new_next_nodes = self.estimate_next_nodes_for_selection_without_caching(
+                        element,
+                        indirect_options.same_type_options.iter(),
+                    )?;
+                    next_nodes.extend(entry.insert(new_next_nodes));
+                }
+            }
+        }
+        for node in &parent_nodes
+            .next_nodes_with_indirect_options
+            .remaining_nodes
+        {
+            match cache.remaining_nodes_to_next_nodes.entry(*node) {
+                Entry::Occupied(entry) => next_nodes.extend(entry.get()),
+                Entry::Vacant(entry) => {
+                    let new_next_nodes = self.estimate_next_nodes_for_selection_without_caching(
+                        element,
+                        std::iter::once(node),
+                    )?;
+                    next_nodes.extend(entry.insert(new_next_nodes));
+                }
+            }
+        }
+        Ok(next_nodes)
+    }
+
+    /// Estimate an upper bound on the next nodes after taking the selection on the given parent
+    /// nodes. Because we're just trying for an upper bound, we assume we can always take
+    /// type-preserving non-collecting transitions, we ignore any conditions on the selection
+    /// edge, and we always type-explode. (We do account for override conditions, which are
+    /// relatively straightforward.)
+    ///
+    /// Since we're iterating through next nodes in the process, for efficiency sake we also
+    /// compute whether there are any reachable cross-subgraph edges from the next nodes
+    /// (without indirect options). This method assumes that inline fragments have type
+    /// conditions.
+    fn estimate_next_nodes_for_selection_without_caching<'c>(
+        &self,
+        element: &OpPathElement,
+        parent_nodes: impl Iterator<Item = &'c NodeIndex>,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let mut next_nodes = IndexSet::default();
+        let nodes_to_object_type_downcasts = &self
+            .parameters
+            .federated_query_graph
+            .non_local_selection_metadata()
+            .nodes_to_object_type_downcasts;
+        match element {
+            OpPathElement::Field(field) => {
+                let Some(field_endpoints) = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .fields_to_endpoints
+                    .get(field.name())
+                else {
+                    return Ok(Default::default());
+                };
+                let mut process_head_node = |node: NodeIndex| {
+                    let Some(target) = field_endpoints.get(&node) else {
+                        return;
+                    };
+                    match target {
+                        FieldTarget::NonOverride(target) => {
+                            next_nodes.insert(*target);
+                        }
+                        FieldTarget::Override(target, condition) => {
+                            if condition.check(&self.parameters.override_conditions) {
+                                next_nodes.insert(*target);
+                            }
+                        }
+                    }
+                };
+                for node in parent_nodes {
+                    // As an upper bound for efficiency sake, we consider both non-type-exploded
+                    // and type-exploded options.
+                    process_head_node(*node);
+                    let Some(object_type_downcasts) = nodes_to_object_type_downcasts.get(node)
+                    else {
+                        continue;
+                    };
+                    match object_type_downcasts {
+                        ObjectTypeDowncasts::NonInterfaceObject(downcasts) => {
+                            for node in downcasts.values() {
+                                process_head_node(*node);
+                            }
+                        }
+                        ObjectTypeDowncasts::InterfaceObject(_) => {
+                            // Interface object fake downcasts only go back to the
+                            // self node, so we ignore them.
+                        }
+                    }
+                }
+            }
+            OpPathElement::InlineFragment(inline_fragment) => {
+                let Some(type_condition_pos) = &inline_fragment.type_condition_position else {
+                    bail!("Inline fragment unexpectedly had no type condition")
+                };
+                let inline_fragment_endpoints = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .inline_fragments_to_endpoints
+                    .get(type_condition_pos.type_name());
+                // If we end up computing runtime types for the type condition, only do it once.
+                let mut possible_runtime_types = None;
+                for node in parent_nodes {
+                    // We check whether there's already a (maybe fake) downcast edge for the
+                    // type condition (note that we've inserted fake downcasts for same-type
+                    // type conditions into the metadata).
+                    if let Some(next_node) = inline_fragment_endpoints.and_then(|e| e.get(node)) {
+                        next_nodes.insert(*next_node);
+                        continue;
+                    }
+
+                    // If not, then we need to type explode across the possible runtime types
+                    // (in the supergraph schema) for the type condition.
+                    let Some(downcasts) = nodes_to_object_type_downcasts.get(node) else {
+                        continue;
+                    };
+                    let possible_runtime_types = match &possible_runtime_types {
+                        Some(possible_runtime_types) => possible_runtime_types,
+                        None => {
+                            let type_condition_in_supergraph_pos = self
+                                .parameters
+                                .supergraph_schema
+                                .get_type(type_condition_pos.type_name().clone())?;
+                            possible_runtime_types.insert(
+                                self.parameters.supergraph_schema.possible_runtime_types(
+                                    type_condition_in_supergraph_pos.try_into()?,
+                                )?,
+                            )
+                        }
+                    };
+
+                    match downcasts {
+                        ObjectTypeDowncasts::NonInterfaceObject(downcasts) => {
+                            for (type_name, target_node) in downcasts {
+                                if possible_runtime_types.contains(&ObjectTypeDefinitionPosition {
+                                    type_name: type_name.clone(),
+                                }) {
+                                    next_nodes.insert(*target_node);
+                                }
+                            }
+                        }
+                        ObjectTypeDowncasts::InterfaceObject(downcasts) => {
+                            for type_name in downcasts {
+                                if possible_runtime_types.contains(&ObjectTypeDefinitionPosition {
+                                    type_name: type_name.clone(),
+                                }) {
+                                    // Note that interface object fake downcasts are self edges,
+                                    // so we're done once we find one.
+                                    next_nodes.insert(*node);
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        self.estimate_nodes_with_indirect_options(next_nodes)
+    }
+
+    /// Estimate the indirect options for the given next nodes, and add them to the given nodes.
+    /// As an upper bound for efficiency's sake, we assume we can take any indirect option (i.e.
+    /// ignore any edge conditions).
+    fn estimate_nodes_with_indirect_options(
+        &self,
+        next_nodes: IndexSet<NodeIndex>,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let mut next_nodes_info = NextNodesInfo {
+            next_nodes,
+            ..Default::default()
+        };
+        for next_node in &next_nodes_info.next_nodes {
+            let next_node_weight = self
+                .parameters
+                .federated_query_graph
+                .node_weight(*next_node)?;
+            next_nodes_info.next_nodes_have_reachable_cross_subgraph_edges = next_nodes_info
+                .next_nodes_have_reachable_cross_subgraph_edges
+                || next_node_weight.has_reachable_cross_subgraph_edges;
+
+            let next_node_type_pos: CompositeTypeDefinitionPosition =
+                next_node_weight.type_.clone().try_into()?;
+            if let Some(options_metadata) = self
+                .parameters
+                .federated_query_graph
+                .non_local_selection_metadata()
+                .types_to_indirect_options
+                .get(next_node_type_pos.type_name())
+            {
+                // If there's an entry in `types_to_indirect_options` for the type, then the
+                // complete digraph for T is non-empty, so we add its type. If it's our first
+                // time seeing this type, we also add any of the complete digraph's interface
+                // object options.
+                if next_nodes_info
+                    .next_nodes_with_indirect_options
+                    .types
+                    .insert(next_node_type_pos.type_name().clone())
+                {
+                    next_nodes_info
+                        .next_nodes_with_indirect_options
+                        .types
+                        .extend(options_metadata.interface_object_options.iter().cloned());
+                }
+                // If the node is a member of the complete digraph, then we don't need to
+                // separately add the remaining node.
+                if options_metadata.same_type_options.contains(next_node) {
+                    continue;
+                }
+            }
+            // We need to add the remaining node, and if its our first time seeing it, we also
+            // add any of its interface object options.
+            if next_nodes_info
+                .next_nodes_with_indirect_options
+                .remaining_nodes
+                .insert(*next_node)
+            {
+                if let Some(options) = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .remaining_nodes_to_interface_object_options
+                    .get(next_node)
+                {
+                    next_nodes_info
+                        .next_nodes_with_indirect_options
+                        .types
+                        .extend(options.iter().cloned());
+                }
+            }
+        }
+
+        Ok(next_nodes_info)
+    }
+}
+
+/// Precompute relevant metadata about the query graph for speeding up the estimation of the
+/// count of non-local selections. Note that none of the algorithms used in this function should
+/// take any longer algorithmically as the rest of query graph creation (and similarly for
+/// query graph memory).
+pub(crate) fn precompute_non_local_selection_metadata(
+    graph: &QueryGraph,
+) -> Result<QueryGraphMetadata, FederationError> {
+    let mut nodes_to_interface_object_options: IndexMap<NodeIndex, IndexSet<Name>> =
+        Default::default();
+    let mut metadata = QueryGraphMetadata::default();
+
+    for edge_ref in graph.graph().edge_references() {
+        match &edge_ref.weight().transition {
+            QueryGraphEdgeTransition::FieldCollection {
+                field_definition_position,
+                ..
+            } => {
+                // We skip selections where the tail is a non-composite type, as we'll never
+                // need to estimate the next nodes for such selections.
+                let Ok(_): Result<CompositeTypeDefinitionPosition, _> = graph
+                    .node_weight(edge_ref.target())?
+                    .type_
+                    .clone()
+                    .try_into()
+                else {
+                    continue;
+                };
+                let target = edge_ref
+                    .weight()
+                    .override_condition
+                    .clone()
+                    .map(|condition| FieldTarget::Override(edge_ref.target(), condition))
+                    .unwrap_or_else(|| FieldTarget::NonOverride(edge_ref.target()));
+                metadata
+                    .fields_to_endpoints
+                    .entry(field_definition_position.field_name().clone())
+                    .or_default()
+                    .insert(edge_ref.source(), target);
+            }
+            QueryGraphEdgeTransition::Downcast {
+                to_type_position, ..
+            } => {
+                if to_type_position.is_object_type() {
+                    let ObjectTypeDowncasts::NonInterfaceObject(downcasts) = metadata
+                        .nodes_to_object_type_downcasts
+                        .entry(edge_ref.source())
+                        .or_insert_with(|| {
+                            ObjectTypeDowncasts::NonInterfaceObject(Default::default())
+                        })
+                    else {
+                        bail!("Unexpectedly found interface object with regular object downcasts")
+                    };
+                    downcasts.insert(to_type_position.type_name().clone(), edge_ref.target());
+                }
+                metadata
+                    .inline_fragments_to_endpoints
+                    .entry(to_type_position.type_name().clone())
+                    .or_default()
+                    .insert(edge_ref.source(), edge_ref.target());
+            }
+            QueryGraphEdgeTransition::InterfaceObjectFakeDownCast { to_type_name, .. } => {
+                // Note that fake downcasts for interface objects are only created to "fake"
+                // object types.
+                let ObjectTypeDowncasts::InterfaceObject(downcasts) = metadata
+                    .nodes_to_object_type_downcasts
+                    .entry(edge_ref.source())
+                    .or_insert_with(|| ObjectTypeDowncasts::InterfaceObject(Default::default()))
+                else {
+                    bail!("Unexpectedly found abstract type with interface object downcasts")
+                };
+                downcasts.insert(to_type_name.clone());
+                metadata
+                    .inline_fragments_to_endpoints
+                    .entry(to_type_name.clone())
+                    .or_default()
+                    .insert(edge_ref.source(), edge_ref.target());
+            }
+            QueryGraphEdgeTransition::KeyResolution
+            | QueryGraphEdgeTransition::RootTypeResolution { .. } => {
+                let head_type_pos: CompositeTypeDefinitionPosition = graph
+                    .node_weight(edge_ref.source())?
+                    .type_
+                    .clone()
+                    .try_into()?;
+                let tail_type_pos: CompositeTypeDefinitionPosition = graph
+                    .node_weight(edge_ref.target())?
+                    .type_
+                    .clone()
+                    .try_into()?;
+                if head_type_pos.type_name() == tail_type_pos.type_name() {
+                    // In this case, we have a non-interface-object key resolution edge or a
+                    // root type resolution edge. The tail must be part of the complete digraph
+                    // for the tail's type, so we record the member.
+                    metadata
+                        .types_to_indirect_options
+                        .entry(tail_type_pos.type_name().clone())
+                        .or_default()
+                        .same_type_options
+                        .insert(edge_ref.target());
+                } else {
+                    // Otherwise, this must be an interface object key resolution edge. We don't
+                    // know the members of the complete digraph for the head's type yet, so we
+                    // can't set the metadata yet, and instead store the head to interface
+                    // object type mapping in a temporary map.
+                    nodes_to_interface_object_options
+                        .entry(edge_ref.source())
+                        .or_default()
+                        .insert(tail_type_pos.type_name().clone());
+                }
+            }
+            QueryGraphEdgeTransition::SubgraphEnteringTransition => {}
+        }
+    }
+
+    // Now that we've finished computing members of the complete digraphs, we can properly track
+    // interface object options.
+    for (node, options) in nodes_to_interface_object_options {
+        let node_type_pos: CompositeTypeDefinitionPosition =
+            graph.node_weight(node)?.type_.clone().try_into()?;
+        if let Some(options_metadata) = metadata
+            .types_to_indirect_options
+            .get_mut(node_type_pos.type_name())
+        {
+            if options_metadata.same_type_options.contains(&node) {
+                options_metadata
+                    .interface_object_options
+                    .extend(options.into_iter());
+                continue;
+            }
+        }
+        metadata
+            .remaining_nodes_to_interface_object_options
+            .insert(node, options);
+    }
+
+    // The interface object options for the complete digraphs are now correct, but we need to
+    // subtract these from any interface object options for remaining nodes.
+    for (node, options) in metadata
+        .remaining_nodes_to_interface_object_options
+        .iter_mut()
+    {
+        let node_type_pos: CompositeTypeDefinitionPosition =
+            graph.node_weight(*node)?.type_.clone().try_into()?;
+        let Some(IndirectOptionsMetadata {
+            interface_object_options,
+            ..
+        }) = metadata
+            .types_to_indirect_options
+            .get(node_type_pos.type_name())
+        else {
+            continue;
+        };
+        options.retain(|type_name| !interface_object_options.contains(type_name));
+    }
+
+    // If this subtraction left any interface object option sets empty, we remove them.
+    metadata
+        .remaining_nodes_to_interface_object_options
+        .retain(|_, options| !options.is_empty());
+
+    // For all composite type nodes, we pretend that there's a self-downcast edge for that type,
+    // as this simplifies next node calculation.
+    for (node, node_weight) in graph.graph().node_references() {
+        let Ok(node_type_pos): Result<CompositeTypeDefinitionPosition, _> =
+            node_weight.type_.clone().try_into()
+        else {
+            continue;
+        };
+        metadata
+            .inline_fragments_to_endpoints
+            .entry(node_type_pos.type_name().clone())
+            .or_default()
+            .insert(node, node);
+        if node_type_pos.is_object_type()
+            && !graph
+                .schema_by_source(&node_weight.source)?
+                .is_interface_object_type(node_type_pos.clone().into())?
+        {
+            let ObjectTypeDowncasts::NonInterfaceObject(downcasts) = metadata
+                .nodes_to_object_type_downcasts
+                .entry(node)
+                .or_insert_with(|| ObjectTypeDowncasts::NonInterfaceObject(Default::default()))
+            else {
+                bail!(
+                    "Unexpectedly found object type with interface object downcasts in supergraph"
+                )
+            };
+            downcasts.insert(node_type_pos.type_name().clone(), node);
+        }
+    }
+
+    // For each subgraph schema, we iterate through its composite types, so that we can collect
+    // metadata relevant to rebasing.
+    for (subgraph_name, subgraph_schema) in graph.subgraph_schemas() {
+        // We pass through each composite type, recording whether the field can be rebased on it
+        // along with interface implements/union membership relationships.
+        let mut fields_to_rebaseable_types: IndexMap<Name, IndexSet<Name>> = Default::default();
+        let mut object_types_to_implementing_composite_types: IndexMap<Name, IndexSet<Name>> =
+            Default::default();
+        let Some(subgraph_metadata) = subgraph_schema.subgraph_metadata() else {
+            bail!("Subgraph schema unexpectedly did not have subgraph metadata")
+        };
+        let from_context_directive_definition_name = &subgraph_metadata
+            .federation_spec_definition()
+            .from_context_directive_definition(subgraph_schema)?
+            .name;
+        for (type_name, type_) in &subgraph_schema.schema().types {
+            match type_ {
+                ExtendedType::Object(type_) => {
+                    // Record fields that don't contain @fromContext as being rebaseable (also
+                    // including __typename).
+                    for (field_name, field_definition) in &type_.fields {
+                        if field_definition.arguments.iter().any(|arg_definition| {
+                            arg_definition
+                                .directives
+                                .has(from_context_directive_definition_name)
+                        }) {
+                            continue;
+                        }
+                        fields_to_rebaseable_types
+                            .entry(field_name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                    // Record the object type as implementing itself.
+                    let implementing_composite_types = object_types_to_implementing_composite_types
+                        .entry(type_name.clone())
+                        .or_default();
+                    implementing_composite_types.insert(type_name.clone());
+                    // For each implements, record the interface type as an implementing type.
+                    if !type_.implements_interfaces.is_empty() {
+                        implementing_composite_types.extend(
+                            type_
+                                .implements_interfaces
+                                .iter()
+                                .map(|interface_name| interface_name.name.clone()),
+                        );
+                    }
+                }
+                ExtendedType::Interface(type_) => {
+                    // Record fields that don't contain @fromContext as being rebaseable (also
+                    // including __typename).
+                    for (field_name, field_definition) in &type_.fields {
+                        if field_definition.arguments.iter().any(|arg_definition| {
+                            arg_definition
+                                .directives
+                                .has(from_context_directive_definition_name)
+                        }) {
+                            continue;
+                        }
+                        fields_to_rebaseable_types
+                            .entry(field_name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                }
+                ExtendedType::Union(type_) => {
+                    // Just record the __typename field as being rebaseable.
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                    // For each member, record the union type as an implementing type.
+                    for member_name in &type_.members {
+                        object_types_to_implementing_composite_types
+                            .entry(member_name.name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                }
+                _ => {}
+            }
+        }
+
+        // With the interface implements/union membership relationships, we can compute which
+        // pairs of types have at least one possible runtime type in their intersection, and
+        // are thus rebaseable.
+        let mut inline_fragments_to_rebaseable_types: IndexMap<Name, IndexSet<Name>> =
+            Default::default();
+        for implementing_types in object_types_to_implementing_composite_types.values() {
+            for type_name in implementing_types {
+                inline_fragments_to_rebaseable_types
+                    .entry(type_name.clone())
+                    .or_default()
+                    .extend(implementing_types.iter().cloned())
+            }
+        }
+
+        // Finally, we can compute the nodes for the rebaseable types, as we'll be working with
+        // those instead of types when checking whether an operation element can be rebased.
+        let types_to_nodes = graph.types_to_nodes_by_source(subgraph_name)?;
+        for (field_name, types) in fields_to_rebaseable_types {
+            metadata
+                .fields_to_rebaseable_parent_nodes
+                .entry(field_name)
+                .or_default()
+                .extend(
+                    types
+                        .iter()
+                        .flat_map(|type_| types_to_nodes.get(type_).map(|nodes| nodes.iter()))
+                        .flatten()
+                        .cloned(),
+                );
+        }
+        for (type_condition_name, types) in inline_fragments_to_rebaseable_types {
+            metadata
+                .inline_fragments_to_rebaseable_parent_nodes
+                .entry(type_condition_name)
+                .or_default()
+                .extend(
+                    types
+                        .iter()
+                        .flat_map(|type_| types_to_nodes.get(type_).map(|nodes| nodes.iter()))
+                        .flatten()
+                        .cloned(),
+                )
+        }
+    }
+    Ok(metadata)
+}
+
+/// During query graph creation, we pre-compute metadata that helps us greatly speed up the
+/// estimation process during request execution. The expected time and memory consumed for
+/// pre-computation is (in the worst case) expected to be on the order of the number of nodes
+/// plus the number of edges.
+///
+/// Note that when the below field docs talk about a "complete digraph", they are referring to
+/// the graph theory concept: https://en.wikipedia.org/wiki/Complete_graph
+#[derive(Debug, Default)]
+pub(crate) struct QueryGraphMetadata {
+    /// When a (resolvable) @key exists on a type T in a subgraph, a key resolution edge is
+    /// created from every subgraph's type T to that subgraph's type T. This similarly holds for
+    /// root type resolution edges. This means that the nodes of type T with such a @key (or are
+    /// operation root types) form a complete digraph in the query graph. These indirect options
+    /// effectively occur as a group in our estimation process, so we track group members here
+    /// per type name, and precompute units of work relative to these groups.
+    ///
+    /// Interface object types I in a subgraph will only sometimes create a key resolution edge
+    /// between an implementing type T in a subgraph and that subgraph's type I. This means the
+    /// nodes of the complete digraph for I are indirect options for such nodes of type T. We
+    /// track any such types I that are reachable for at least one node in the complete digraph
+    /// for type T here as well.
+    types_to_indirect_options: IndexMap<Name, IndirectOptionsMetadata>,
+    /// For nodes of a type T that aren't in their complete digraph (due to not having a @key),
+    /// these remaining nodes will have the complete digraph of T (and any interface object
+    /// complete digraphs) as indirect options, but these remaining nodes may separately have
+    /// more indirect options that are not options for the complete digraph of T, specifically
+    /// if the complete digraph for T has no key resolution edges to an interface object I, but
+    /// this remaining node does. We keep track of such interface object types for those
+    /// remaining nodes here.
+    remaining_nodes_to_interface_object_options: IndexMap<NodeIndex, IndexSet<Name>>,
+    /// A map of field names to the endpoints of field query graph edges with that field name. Note
+    /// we additionally store the progressive overrides label, if the edge is conditioned on it.
+    fields_to_endpoints: IndexMap<Name, IndexMap<NodeIndex, FieldTarget>>,
+    /// A map of type condition names to endpoints of downcast query graph edges with that type
+    /// condition name, including fake downcasts for interface objects, and a non-existent edge that
+    /// represents a type condition name equal to the parent type.
+    inline_fragments_to_endpoints: IndexMap<Name, IndexMap<NodeIndex, NodeIndex>>,
+    /// A map of composite type nodes to their downcast edges that lead specifically to an object
+    /// type (i.e., the possible runtime types of the node's type).
+    nodes_to_object_type_downcasts: IndexMap<NodeIndex, ObjectTypeDowncasts>,
+    /// A map of field names to parent nodes whose corresponding type and schema can be rebased on
+    /// by the field.
+    fields_to_rebaseable_parent_nodes: IndexMap<Name, IndexSet<NodeIndex>>,
+    /// A map of type condition names to parent nodes whose corresponding type and schema can be
+    /// rebased on by an inline fragment with that type condition.
+    inline_fragments_to_rebaseable_parent_nodes: IndexMap<Name, IndexSet<NodeIndex>>,
+}
+
+/// Indirect option metadata for the complete digraph for type T. See [QueryGraphMetadata] for
+/// more information about how we group indirect options into complete digraphs.
+#[derive(Debug, Default)]
+pub(crate) struct IndirectOptionsMetadata {
+    /// The members of the complete digraph for type T.
+    same_type_options: IndexSet<NodeIndex>,
+    /// Any interface object types I that are reachable for at least one node in the complete
+    /// digraph for type T.
+    interface_object_options: IndexSet<Name>,
+}
+
+#[derive(Debug)]
+enum FieldTarget {
+    /// Normal non-overridden fields, which don't have a label condition.
+    NonOverride(NodeIndex),
+    /// Overridden fields, which have a label condition.
+    Override(NodeIndex, OverrideCondition),
+}
+
+#[derive(Debug)]
+enum ObjectTypeDowncasts {
+    /// Normal non-interface-object types have regular downcasts to their object type nodes.
+    NonInterfaceObject(IndexMap<Name, NodeIndex>),
+    /// Interface object types have "fake" downcasts to nodes that are really the self node.
+    InterfaceObject(IndexSet<Name>),
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct State {
+    /// An estimation of the number of non-local selections for the whole operation (where the count
+    /// for a given selection set is scaled by the number of tail nodes at that selection set). Note
+    /// this does not count selections from recursive query planning.
+    pub(crate) count: u64,
+    /// Whenever we take a selection on a set of nodes with indirect options, we cache the
+    /// resulting nodes here.
+    next_nodes_cache: IndexMap<SelectionKey, NextNodesCache>,
+}
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+enum SelectionKey {
+    /// For field selections, this is the field's name.
+    Field(Name),
+    /// For inline fragment selections, this is the type condition's name.
+    InlineFragment(Name),
+}
+
+/// See [QueryGraphMetadata] for more information about how we group indirect options into
+/// complete digraphs.
+#[derive(Debug, Default)]
+struct NextNodesCache {
+    /// This is the merged next node info for selections on the set of nodes in the complete
+    /// digraph for the given type T. Note that this does not merge in the next node info for
+    /// any interface object options reachable from nodes in that complete digraph for T.
+    types_to_next_nodes: IndexMap<Name, NextNodesInfo>,
+    /// This is the next node info for selections on the given node. Note that this does not
+    /// merge in the next node info for any interface object options reachable from that node.
+    remaining_nodes_to_next_nodes: IndexMap<NodeIndex, NextNodesInfo>,
+}
+
+#[derive(Clone, Debug, Default)]
+struct NextNodesInfo {
+    /// The next nodes after taking the selection.
+    next_nodes: IndexSet<NodeIndex>,
+    /// Whether any cross-subgraph edges are reachable from any next nodes.
+    next_nodes_have_reachable_cross_subgraph_edges: bool,
+    /// These are the next nodes along with indirect options, represented succinctly by the
+    /// types of any complete digraphs along with remaining nodes.
+    next_nodes_with_indirect_options: NodesWithIndirectOptionsInfo,
+}
+
+impl NextNodesInfo {
+    fn extend(&mut self, other: &Self) {
+        self.next_nodes.extend(other.next_nodes.iter().cloned());
+        self.next_nodes_have_reachable_cross_subgraph_edges = self
+            .next_nodes_have_reachable_cross_subgraph_edges
+            || other.next_nodes_have_reachable_cross_subgraph_edges;
+        self.next_nodes_with_indirect_options
+            .extend(&other.next_nodes_with_indirect_options);
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+struct NodesWithIndirectOptionsInfo {
+    /// For indirect options that are representable as complete digraphs for a type T, these are
+    /// those types.
+    types: IndexSet<Name>,
+    /// For any nodes of type T that aren't in their complete digraphs for type T, these are
+    /// those nodes.
+    remaining_nodes: IndexSet<NodeIndex>,
+}
+
+impl NodesWithIndirectOptionsInfo {
+    fn extend(&mut self, other: &Self) {
+        self.types.extend(other.types.iter().cloned());
+        self.remaining_nodes
+            .extend(other.remaining_nodes.iter().cloned());
+    }
+}

@@ -1,6 +1,7 @@
 use std::ops::ControlFlow;
 use std::sync::Arc;
 
+use apollo_compiler::Name;
 use apollo_compiler::collections::IndexSet;
 use petgraph::graph::EdgeIndex;
 use petgraph::graph::NodeIndex;
@@ -45,13 +46,14 @@ use crate::query_plan::query_planner::QueryPlannerConfig;
 use crate::query_plan::query_planner::QueryPlanningStatistics;
 use crate::query_plan::query_planner::compute_root_fetch_groups;
 use crate::schema::ValidFederationSchema;
-use crate::schema::position::AbstractTypeDefinitionPosition;
 use crate::schema::position::CompositeTypeDefinitionPosition;
 use crate::schema::position::ObjectTypeDefinitionPosition;
 use crate::schema::position::SchemaRootDefinitionKind;
 use crate::utils::logging::format_open_branch;
 use crate::utils::logging::snapshot;
 
+pub(crate) mod non_local_selections_estimation;
+
 #[cfg(feature = "snapshot_tracing")]
 mod snapshot_helper {
     // A module to import functions only used within `snapshot!(...)` macros.
@@ -82,8 +84,7 @@ pub(crate) struct QueryPlanningParameters<'a> {
     /// subgraphs.
     // PORT_NOTE: Named `inconsistentAbstractTypesRuntimes` in the JS codebase, which was slightly
     // confusing.
-    pub(crate) abstract_types_with_inconsistent_runtime_types:
-        Arc<IndexSet<AbstractTypeDefinitionPosition>>,
+    pub(crate) abstract_types_with_inconsistent_runtime_types: Arc<IndexSet<Name>>,
     /// The configuration for the query planner.
     pub(crate) config: QueryPlannerConfig,
     pub(crate) statistics: &'a QueryPlanningStatistics,
@@ -248,6 +249,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
         has_defers: bool,
         root_kind: SchemaRootDefinitionKind,
         cost_processor: FetchDependencyGraphToCostProcessor,
+        non_local_selection_state: Option<&mut non_local_selections_estimation::State>,
     ) -> Result<Self, FederationError> {
         Self::new_inner(
             parameters,
@@ -256,6 +258,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
             parameters.fetch_id_generator.clone(),
             root_kind,
             cost_processor,
+            non_local_selection_state,
             Default::default(),
             Default::default(),
             Default::default(),
@@ -275,6 +278,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
         id_generator: Arc<FetchIdGenerator>,
         root_kind: SchemaRootDefinitionKind,
         cost_processor: FetchDependencyGraphToCostProcessor,
+        non_local_selection_state: Option<&mut non_local_selections_estimation::State>,
         initial_context: OpGraphPathContext,
         excluded_destinations: ExcludedDestinations,
         excluded_conditions: ExcludedConditions,
@@ -332,6 +336,20 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
 
         traversal.open_branches = map_options_to_selections(selection_set, initial_options);
 
+        if let Some(non_local_selection_state) = non_local_selection_state {
+            if traversal
+                .check_non_local_selections_limit_exceeded_at_root(non_local_selection_state)?
+            {
+                return Err(SingleFederationError::QueryPlanComplexityExceeded {
+                    message: format!(
+                        "Number of non-local selections exceeds limit of {}",
+                        Self::MAX_NON_LOCAL_SELECTIONS,
+                    ),
+                }
+                .into());
+            }
+        }
+
         Ok(traversal)
     }
 
@@ -639,8 +657,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
                             Some(type_condition) => self
                                 .parameters
                                 .abstract_types_with_inconsistent_runtime_types
-                                .iter()
-                                .any(|ty| ty.type_name() == type_condition.type_name()),
+                                .contains(type_condition.type_name()),
                             None => false,
                         }
                     }
@@ -1165,6 +1182,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
             self.id_generator.clone(),
             self.root_kind,
             self.cost_processor,
+            None,
             context.clone(),
             excluded_destinations.clone(),
             excluded_conditions.add_item(edge_conditions),

@@ -44,24 +44,24 @@ fn valid_query_plan() {
     ";
 
     // Number of bytes when the heap size reached its global maximum with a 5% buffer.
-    // Actual number: 683_609.
-    const MAX_BYTES_QUERY_PLANNER: usize = 718_000; // ~720 KiB
+    // Actual number: 744_494.
+    const MAX_BYTES_QUERY_PLANNER: usize = 781_718; // ~763 KiB
 
     // Total number of allocations with a 5% buffer.
-    // Actual number: 13_891.
-    const MAX_ALLOCATIONS_QUERY_PLANNER: u64 = 14_600;
+    // Actual number: 15_403.
+    const MAX_ALLOCATIONS_QUERY_PLANNER: u64 = 16_173;
 
     // Number of bytes when the heap size reached its global maximum with a 5% buffer.
-    // Actual number: 816_807.
+    // Actual number: 864_783.
     //
-    // Planning adds 133_198 bytes to heap max (816_807-683_609=133_198).
-    const MAX_BYTES_QUERY_PLAN: usize = 857_000; // ~860 KiB
+    // Planning adds 120_289 bytes to heap max (864_783-744_494=120_289).
+    const MAX_BYTES_QUERY_PLAN: usize = 908_022; // ~886 KiB
 
     // Total number of allocations with a 5% buffer.
-    // Actual number: 21_277.
+    // Actual number: 22_937.
     //
-    // Planning adds 7_386 allocations (21_277-13_891=7_386).
-    const MAX_ALLOCATIONS_QUERY_PLAN: u64 = 22_400;
+    // Planning adds 7_534 allocations (22_937-15_403=7_534).
+    const MAX_ALLOCATIONS_QUERY_PLAN: u64 = 24_083;
 
     let schema = std::fs::read_to_string(SCHEMA).unwrap();
 

@@ -26,6 +26,7 @@ fn it_handles_progressive_override_on_root_fields() {
         QueryPlanOptions {
             override_conditions: vec!["test".to_string()],
             check_for_cooperative_cancellation: None,
+            ..Default::default()
         },
         @r###"
         QueryPlan {
@@ -120,6 +121,7 @@ fn it_handles_progressive_override_on_entity_fields() {
         QueryPlanOptions {
             override_conditions: vec!["test".to_string()],
             check_for_cooperative_cancellation: None,
+            ..Default::default()
         },
         @r###"
           QueryPlan {
@@ -280,6 +282,7 @@ fn it_handles_progressive_override_on_nested_entity_fields() {
         QueryPlanOptions {
             override_conditions: vec!["test".to_string()],
             check_for_cooperative_cancellation: None,
+            ..Default::default()
         },
         @r###"
           QueryPlan {

@@ -47,6 +47,7 @@ fn it_overrides_to_s2_when_label_is_provided() {
         QueryPlanOptions {
             override_conditions: vec!["test".to_string()],
             check_for_cooperative_cancellation: None,
+            ..Default::default()
         },
         @r###"
           QueryPlan {
@@ -160,6 +161,7 @@ fn it_overrides_f1_to_s3_when_label_is_provided() {
         QueryPlanOptions {
             override_conditions: vec!["test".to_string()],
             check_for_cooperative_cancellation: None,
+            ..Default::default()
         },
         @r###"
           QueryPlan {

@@ -478,6 +478,15 @@ impl InstrumentData {
             "opt.apollo.dev".to_string(),
             atomic_is_true(&crate::executable::APOLLO_ROUTER_DEV_MODE),
         );
+        attributes.insert(
+            "opt.security.recursive_selections".to_string(),
+            crate::services::layers::query_analysis::recursive_selections_check_enabled().into(),
+        );
+        attributes.insert(
+            "opt.security.non_local_selections".to_string(),
+            crate::query_planner::query_planner_service::non_local_selections_check_enabled()
+                .into(),
+        );
 
         self.data
             .insert("apollo.router.config.env".to_string(), (1, attributes));

@@ -1,6 +1,7 @@
 ---
 source: apollo-router/src/configuration/metrics.rs
-expression: "&metrics.non_zero()"
+expression: "& metrics.non_zero()"
+snapshot_kind: text
 ---
 - name: apollo.router.config.env
   data:
@@ -14,4 +15,5 @@ expression: "&metrics.non_zero()"
           opt.apollo.license.path: true
           opt.apollo.supergraph.path: true
           opt.apollo.supergraph.urls: true
-
+          opt.security.non_local_selections: true
+          opt.security.recursive_selections: true

@@ -3,6 +3,7 @@
 use std::fmt::Debug;
 use std::ops::ControlFlow;
 use std::sync::Arc;
+use std::sync::OnceLock;
 use std::task::Poll;
 use std::time::Instant;
 
@@ -59,6 +60,25 @@ pub(crate) const RUST_QP_MODE: &str = "rust";
 const UNSUPPORTED_FED1: &str = "fed1";
 const INTERNAL_INIT_ERROR: &str = "internal";
 
+const ENV_DISABLE_NON_LOCAL_SELECTIONS_CHECK: &str =
+    "APOLLO_ROUTER_DISABLE_SECURITY_NON_LOCAL_SELECTIONS_CHECK";
+/// Should we enforce the non-local selections limit? Default true, can be toggled off with an
+/// environment variable.
+///
+/// Disabling this check is very much not advisable and we don't expect that anyone will need to do
+/// it. In the extremely unlikely case that the new protection breaks someone's legitimate queries,
+/// though, they could temporarily disable this individual limit so they can still benefit from the
+/// other new limits, until we improve the detection.
+pub(crate) fn non_local_selections_check_enabled() -> bool {
+    static ON: OnceLock<bool> = OnceLock::new();
+    *ON.get_or_init(|| {
+        let disabled =
+            std::env::var(ENV_DISABLE_NON_LOCAL_SELECTIONS_CHECK).as_deref() == Ok("true");
+
+        !disabled
+    })
+}
+
 /// A query planner that calls out to the apollo-federation crate.
 ///
 /// No caching is performed. To cache, wrap in a [`CachingQueryPlanner`].
@@ -139,6 +159,7 @@ impl QueryPlannerService {
             let query_plan_options = QueryPlanOptions {
                 override_conditions: plan_options.override_conditions,
                 check_for_cooperative_cancellation: Some(&check),
+                non_local_selections_limit_enabled: non_local_selections_check_enabled(),
             };
 
             let result = operation

@@ -6,11 +6,16 @@ use std::fmt::Display;
 use std::fmt::Formatter;
 use std::hash::Hash;
 use std::sync::Arc;
+use std::sync::OnceLock;
 
 use apollo_compiler::ExecutableDocument;
+use apollo_compiler::Name;
 use apollo_compiler::Node;
 use apollo_compiler::ast;
 use apollo_compiler::executable::Operation;
+use apollo_compiler::executable::Selection;
+use apollo_compiler::executable::SelectionSet;
+use apollo_compiler::response::GraphQLError;
 use apollo_compiler::validation::Valid;
 use http::StatusCode;
 use lru::LruCache;
@@ -25,6 +30,7 @@ use crate::compute_job;
 use crate::compute_job::MaybeBackPressureError;
 use crate::context::OPERATION_KIND;
 use crate::context::OPERATION_NAME;
+use crate::error::ValidationErrors;
 use crate::graphql::Error;
 use crate::graphql::ErrorExtension;
 use crate::graphql::IntoGraphQLErrors;
@@ -41,6 +47,25 @@ use crate::spec::QueryHash;
 use crate::spec::Schema;
 use crate::spec::SpecError;
 
+const ENV_DISABLE_RECURSIVE_SELECTIONS_CHECK: &str =
+    "APOLLO_ROUTER_DISABLE_SECURITY_RECURSIVE_SELECTIONS_CHECK";
+/// Should we enforce the recursive selections limit? Default true, can be toggled off with an
+/// environment variable.
+///
+/// Disabling this check is very much not advisable and we don't expect that anyone will need to do
+/// it. In the extremely unlikely case that the new protection breaks someone's legitimate queries,
+/// though, they could temporarily disable this individual limit so they can still benefit from the
+/// other new limits, until we improve the detection.
+pub(crate) fn recursive_selections_check_enabled() -> bool {
+    static ON: OnceLock<bool> = OnceLock::new();
+    *ON.get_or_init(|| {
+        let disabled =
+            std::env::var(ENV_DISABLE_RECURSIVE_SELECTIONS_CHECK).as_deref() == Ok("true");
+
+        !disabled
+    })
+}
+
 /// A layer-like type that handles several aspects of query parsing and analysis.
 ///
 /// The supergraph layer implementation is in [QueryAnalysisLayer::supergraph_request].
@@ -61,6 +86,8 @@ struct QueryAnalysisKey {
 }
 
 impl QueryAnalysisLayer {
+    const MAX_RECURSIVE_SELECTIONS: u32 = 10_000_000;
+
     pub(crate) async fn new(schema: Arc<Schema>, configuration: Arc<Configuration>) -> Self {
         let enable_authorization_directives =
             AuthorizationPlugin::enable_directives(&configuration, &schema).unwrap_or(false);
@@ -110,6 +137,34 @@ impl QueryAnalysisLayer {
                     schema.as_ref(),
                     conf.as_ref(),
                 )
+                .and_then(|doc| {
+                    let recursive_selections = Self::count_recursive_selections(
+                        &doc.executable,
+                        &mut Default::default(),
+                        &doc.operation.selection_set,
+                        0,
+                    );
+                    if recursive_selections.is_none() {
+                        if recursive_selections_check_enabled() {
+                            return Err(SpecError::ValidationError(ValidationErrors {
+                                errors: vec![GraphQLError {
+                                    message:
+                                        "Maximum recursive selections limit exceeded in this operation"
+                                            .to_string(),
+                                    locations: Default::default(),
+                                    path: Default::default(),
+                                    extensions: Default::default(),
+                                }],
+                            }))
+                        }
+                        tracing::info!(
+                            operation_name = ?operation_name,
+                            limit = Self::MAX_RECURSIVE_SELECTIONS,
+                            "operation exceeded maximum recursive selections limit, but limit is forcefully disabled",
+                        );
+                    }
+                    Ok(doc)
+                })
             })
         })
         .map_err(MaybeBackPressureError::TemporaryError)?
@@ -123,6 +178,66 @@ impl QueryAnalysisLayer {
         .map_err(MaybeBackPressureError::PermanentError)
     }
 
+    /// Measure the number of selections that would be encountered if we walked the given selection
+    /// set while recursing into fragment spreads, and add it to the given count. `None` is returned
+    /// instead if this number exceeds `Self::MAX_RECURSIVE_SELECTIONS`.
+    ///
+    /// This function assumes that fragments referenced by spreads exist and that they don't form
+    /// cycles. If a fragment spread appears multiple times for the same named fragment, it is
+    /// counted multiple times.
+    fn count_recursive_selections<'a>(
+        document: &'a Valid<ExecutableDocument>,
+        fragment_cache: &mut HashMap<&'a Name, u32>,
+        selection_set: &'a SelectionSet,
+        mut count: u32,
+    ) -> Option<u32> {
+        for selection in &selection_set.selections {
+            count = count
+                .checked_add(1)
+                .take_if(|v| *v <= Self::MAX_RECURSIVE_SELECTIONS)?;
+            match selection {
+                Selection::Field(field) => {
+                    count = Self::count_recursive_selections(
+                        document,
+                        fragment_cache,
+                        &field.selection_set,
+                        count,
+                    )?;
+                }
+                Selection::InlineFragment(fragment) => {
+                    count = Self::count_recursive_selections(
+                        document,
+                        fragment_cache,
+                        &fragment.selection_set,
+                        count,
+                    )?;
+                }
+                Selection::FragmentSpread(fragment) => {
+                    let name = &fragment.fragment_name;
+                    if let Some(cached) = fragment_cache.get(name) {
+                        count = count
+                            .checked_add(*cached)
+                            .take_if(|v| *v <= Self::MAX_RECURSIVE_SELECTIONS)?;
+                    } else {
+                        let old_count = count;
+                        count = Self::count_recursive_selections(
+                            document,
+                            fragment_cache,
+                            &document
+                                .fragments
+                                .get(&fragment.fragment_name)
+                                .expect("validation should have ensured referenced fragments exist")
+                                .selection_set,
+                            count,
+                        )?;
+                        fragment_cache.insert(name, count - old_count);
+                    };
+                }
+            }
+        }
+        Some(count)
+    }
+
     /// Parses the GraphQL in the supergraph request and computes Apollo usage references.
     ///
     /// This functions similarly to a checkpoint service, short-circuiting the pipeline on error

@@ -139,29 +139,29 @@ fn count<'a>(
         match selection {
             executable::Selection::Field(field) => {
                 let nested = count(document, fragment_cache, &field.selection_set);
-                counts.depth = counts.depth.max(1 + nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.depth = counts.depth.max(nested.depth.saturating_add(1));
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
                 // Multiple aliases for the same field could use different arguments
                 // Until we do full merging for limit checking purpose,
                 // approximate measured height with an upper bound rather than a lower bound.
                 let used_name = if let Some(alias) = &field.alias {
-                    counts.aliases += 1;
+                    counts.aliases = counts.aliases.saturating_add(1);
                     alias
                 } else {
                     &field.name
                 };
                 let not_seen_before = fields_seen.insert(used_name);
                 if not_seen_before {
-                    counts.height += 1;
-                    counts.root_fields += 1;
+                    counts.height = counts.height.saturating_add(1);
+                    counts.root_fields = counts.root_fields.saturating_add(1);
                 }
             }
             executable::Selection::InlineFragment(fragment) => {
                 let nested = count(document, fragment_cache, &fragment.selection_set);
                 counts.depth = counts.depth.max(nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
             }
             executable::Selection::FragmentSpread(fragment) => {
                 let name = &fragment.fragment_name;
@@ -190,8 +190,8 @@ fn count<'a>(
                     Some(Computation::Done(cached)) => nested = *cached,
                 }
                 counts.depth = counts.depth.max(nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
             }
         }
     }

@@ -0,0 +1,7 @@
+### Certain query patterns may cause resource exhaustion
+
+Corrects a set of denial-of-service (DOS) vulnerabilities that made it possible for an attacker to render router inoperable with certain simple query patterns due to uncontrolled resource consumption. All prior-released versions and configurations are vulnerable except those where `persisted_queries.enabled`, `persisted_queries.safelist.enabled`, and `persisted_queries.safelist.require_id` are all `true`.
+
+See the associated GitHub Advisories [GHSA-3j43-9v8v-cp3f](https://github.com/apollographql/router/security/advisories/GHSA-3j43-9v8v-cp3f), [GHSA-84m6-5m72-45fp](https://github.com/apollographql/router/security/advisories/GHSA-84m6-5m72-45fp), [GHSA-75m2-jhh5-j5g2](https://github.com/apollographql/router/security/advisories/GHSA-75m2-jhh5-j5g2), and [GHSA-94hh-jmq8-2fgp](https://github.com/apollographql/router/security/advisories/GHSA-94hh-jmq8-2fgp), and the `apollo-compiler` GitHub Advisory [GHSA-7mpv-9xg6-5r79](https://github.com/apollographql/apollo-rs/security/advisories/GHSA-7mpv-9xg6-5r79) for more information.
+
+By [@sachindshinde](https://github.com/sachindshinde) and [@goto-bus-stop](https://github.com/goto-bus-stop).

@@ -32,6 +32,7 @@ use crate::query_graph::QueryGraphEdge;
 use crate::query_graph::QueryGraphEdgeTransition;
 use crate::query_graph::QueryGraphNode;
 use crate::query_graph::QueryGraphNodeType;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation::precompute_non_local_selection_metadata;
 use crate::schema::ValidFederationSchema;
 use crate::schema::field_set::parse_field_set;
 use crate::schema::position::AbstractTypeDefinitionPosition;
@@ -77,6 +78,7 @@ pub fn build_federated_query_graph(
         root_kinds_to_nodes_by_source: Default::default(),
         non_trivial_followup_edges: Default::default(),
         arguments_to_context_ids_by_source: Default::default(),
+        non_local_selection_metadata: Default::default(),
     };
     let query_graph =
         extract_subgraphs_from_supergraph(&supergraph_schema, validate_extracted_subgraphs)?
@@ -112,6 +114,7 @@ pub fn build_query_graph(
         root_kinds_to_nodes_by_source: Default::default(),
         non_trivial_followup_edges: Default::default(),
         arguments_to_context_ids_by_source: Default::default(),
+        non_local_selection_metadata: Default::default(),
     };
     let builder = SchemaQueryGraphBuilder::new(query_graph, name, schema, None, false)?;
     query_graph = builder.build()?;
@@ -1002,6 +1005,10 @@ impl FederatedQueryGraphBuilder {
         self.handle_interface_object()?;
         // This method adds no nodes/edges, but just precomputes followup edge information.
         self.precompute_non_trivial_followup_edges()?;
+        // This method adds no nodes/edges, but just precomputes metadata for estimating the count
+        // of non_local_selections.
+        self.base.query_graph.non_local_selection_metadata =
+            precompute_non_local_selection_metadata(&self.base.query_graph)?;
         Ok(self.base.build())
     }
 

@@ -483,6 +483,15 @@ impl OpPathElement {
         }
     }
 
+    pub(crate) fn has_defer(&self) -> bool {
+        match self {
+            OpPathElement::Field(_) => false,
+            OpPathElement::InlineFragment(inline_fragment) => {
+                inline_fragment.directives.has("defer")
+            }
+        }
+    }
+
     /// Returns this fragment element but with any @defer directive on it removed.
     ///
     /// This method will return `None` if, upon removing @defer, the fragment has no conditions nor

@@ -50,6 +50,7 @@ use crate::query_graph::graph_path::OpGraphPathTrigger;
 use crate::query_graph::graph_path::OpPathElement;
 use crate::query_plan::QueryPlanCost;
 use crate::query_plan::query_planner::EnabledOverrideConditions;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub(crate) struct QueryGraphNode {
@@ -201,7 +202,7 @@ impl QueryGraphEdge {
         conditions_to_check: &EnabledOverrideConditions,
     ) -> bool {
         if let Some(override_condition) = &self.override_condition {
-            override_condition.condition == conditions_to_check.contains(&override_condition.label)
+            override_condition.check(conditions_to_check)
         } else {
             true
         }
@@ -238,6 +239,12 @@ pub(crate) struct OverrideCondition {
     pub(crate) condition: bool,
 }
 
+impl OverrideCondition {
+    pub(crate) fn check(&self, enabled_conditions: &EnabledOverrideConditions) -> bool {
+        self.condition == enabled_conditions.contains(&self.label)
+    }
+}
+
 impl Display for OverrideCondition {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(f, "{} = {}", self.label, self.condition)
@@ -405,6 +412,9 @@ pub struct QueryGraph {
     /// argument coordinates). This identifier is called the "context ID".
     arguments_to_context_ids_by_source:
         IndexMap<Arc<str>, IndexMap<ObjectFieldArgumentDefinitionPosition, Name>>,
+    /// To speed up the estimation of counting non-local selections, we precompute specific metadata
+    /// about the query graph and store that here.
+    non_local_selection_metadata: non_local_selections_estimation::QueryGraphMetadata,
 }
 
 impl QueryGraph {
@@ -500,7 +510,7 @@ impl QueryGraph {
         self.types_to_nodes_by_source(&self.current_source)
     }
 
-    fn types_to_nodes_by_source(
+    pub(super) fn types_to_nodes_by_source(
         &self,
         source: &str,
     ) -> Result<&IndexMap<NamedType, IndexSet<NodeIndex>>, FederationError> {
@@ -582,6 +592,12 @@ impl QueryGraph {
         !self.arguments_to_context_ids_by_source.is_empty()
     }
 
+    pub(crate) fn non_local_selection_metadata(
+        &self,
+    ) -> &non_local_selections_estimation::QueryGraphMetadata {
+        &self.non_local_selection_metadata
+    }
+
     /// All outward edges from the given node (including self-key and self-root-type-resolution
     /// edges). Primarily used by `@defer`, when needing to re-enter a subgraph for a deferred
     /// section.

@@ -42,6 +42,7 @@ use crate::query_plan::query_planning_traversal::BestQueryPlanInfo;
 use crate::query_plan::query_planning_traversal::QueryPlanningParameters;
 use crate::query_plan::query_planning_traversal::QueryPlanningTraversal;
 use crate::query_plan::query_planning_traversal::convert_type_from_subgraph;
+use crate::query_plan::query_planning_traversal::non_local_selections_estimation;
 use crate::schema::ValidFederationSchema;
 use crate::schema::position::AbstractTypeDefinitionPosition;
 use crate::schema::position::CompositeTypeDefinitionPosition;
@@ -170,7 +171,7 @@ pub struct QueryPlanningStatistics {
     pub evaluated_plan_paths: Cell<usize>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Clone)]
 pub struct QueryPlanOptions {
     /// A set of labels which will be used _during query planning_ to
     /// enable/disable edges with a matching label in their override condition.
@@ -179,6 +180,18 @@ pub struct QueryPlanOptions {
     /// progressive @override feature.
     // PORT_NOTE: In JS implementation this was a Map
     pub override_conditions: Vec<String>,
+    /// Impose a limit on the number of non-local selections, which can be a
+    /// performance hazard. On by default.
+    pub non_local_selections_limit_enabled: bool,
+}
+
+impl Default for QueryPlanOptions {
+    fn default() -> Self {
+        Self {
+            override_conditions: Vec::new(),
+            non_local_selections_limit_enabled: true,
+        }
+    }
 }
 
 #[derive(Debug, Default, Clone)]
@@ -204,7 +217,7 @@ pub struct QueryPlanner {
     /// subgraphs.
     // PORT_NOTE: Named `inconsistentAbstractTypesRuntimes` in the JS codebase, which was slightly
     // confusing.
-    abstract_types_with_inconsistent_runtime_types: IndexSet<AbstractTypeDefinitionPosition>,
+    abstract_types_with_inconsistent_runtime_types: IndexSet<Name>,
 }
 
 impl QueryPlanner {
@@ -297,6 +310,7 @@ impl QueryPlanner {
             .get_types()
             .filter_map(|position| AbstractTypeDefinitionPosition::try_from(position).ok())
             .filter(|position| is_inconsistent(position.clone()))
+            .map(|position| position.type_name().clone())
             .collect::<IndexSet<_>>();
 
         Ok(Self {
@@ -417,10 +431,23 @@ impl QueryPlanner {
             fetch_id_generator: Arc::new(FetchIdGenerator::new()),
         };
 
+        let mut non_local_selection_state = options
+            .non_local_selections_limit_enabled
+            .then(non_local_selections_estimation::State::default);
         let root_node = if !defer_conditions.is_empty() {
-            compute_plan_for_defer_conditionals(&mut parameters, &mut processor, defer_conditions)
+            compute_plan_for_defer_conditionals(
+                &mut parameters,
+                &mut processor,
+                defer_conditions,
+                &mut non_local_selection_state,
+            )
         } else {
-            compute_plan_internal(&mut parameters, &mut processor, has_defers)
+            compute_plan_internal(
+                &mut parameters,
+                &mut processor,
+                has_defers,
+                &mut non_local_selection_state,
+            )
         }?;
 
         let root_node = match root_node {
@@ -495,6 +522,7 @@ impl QueryPlanner {
 fn compute_root_serial_dependency_graph(
     parameters: &QueryPlanningParameters,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Vec<FetchDependencyGraph>, FederationError> {
     let QueryPlanningParameters {
         supergraph_schema,
@@ -521,14 +549,24 @@ fn compute_root_serial_dependency_graph(
         mut fetch_dependency_graph,
         path_tree: mut prev_path,
         ..
-    } = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+    } = compute_root_parallel_best_plan(
+        parameters,
+        selection_set,
+        has_defers,
+        non_local_selection_state,
+    )?;
     let mut prev_subgraph = only_root_subgraph(&fetch_dependency_graph)?;
     for selection_set in split_roots {
         let BestQueryPlanInfo {
             fetch_dependency_graph: new_dep_graph,
             path_tree: new_path,
             ..
-        } = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+        } = compute_root_parallel_best_plan(
+            parameters,
+            selection_set,
+            has_defers,
+            non_local_selection_state,
+        )?;
         let new_subgraph = only_root_subgraph(&new_dep_graph)?;
         if new_subgraph == prev_subgraph {
             // The new operation (think 'mutation' operation) is on the same subgraph than the previous one, so we can concat them in a single fetch
@@ -648,10 +686,16 @@ pub(crate) fn compute_root_fetch_groups(
 fn compute_root_parallel_dependency_graph(
     parameters: &QueryPlanningParameters,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<FetchDependencyGraph, FederationError> {
     trace!("Starting process to construct a parallel fetch dependency graph");
     let selection_set = parameters.operation.selection_set.clone();
-    let best_plan = compute_root_parallel_best_plan(parameters, selection_set, has_defers)?;
+    let best_plan = compute_root_parallel_best_plan(
+        parameters,
+        selection_set,
+        has_defers,
+        non_local_selection_state,
+    )?;
     snapshot!(
         "FetchDependencyGraph",
         best_plan.fetch_dependency_graph.to_dot(),
@@ -664,13 +708,15 @@ fn compute_root_parallel_best_plan(
     parameters: &QueryPlanningParameters,
     selection: SelectionSet,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<BestQueryPlanInfo, FederationError> {
     let planning_traversal = QueryPlanningTraversal::new(
         parameters,
         selection,
         has_defers,
         parameters.operation.root_kind,
         FetchDependencyGraphToCostProcessor,
+        non_local_selection_state.as_mut(),
     )?;
 
     // Getting no plan means the query is essentially unsatisfiable (it's a valid query, but we can prove it will never return a result),
@@ -684,11 +730,16 @@ fn compute_plan_internal(
     parameters: &mut QueryPlanningParameters,
     processor: &mut FetchDependencyGraphToQueryPlanProcessor,
     has_defers: bool,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Option<PlanNode>, FederationError> {
     let root_kind = parameters.operation.root_kind;
 
     let (main, deferred, primary_selection) = if root_kind == SchemaRootDefinitionKind::Mutation {
-        let dependency_graphs = compute_root_serial_dependency_graph(parameters, has_defers)?;
+        let dependency_graphs = compute_root_serial_dependency_graph(
+            parameters,
+            has_defers,
+            non_local_selection_state,
+        )?;
         let mut main = None;
         let mut deferred = vec![];
         let mut primary_selection = None::<SelectionSet>;
@@ -712,7 +763,11 @@ fn compute_plan_internal(
         }
         (main, deferred, primary_selection)
     } else {
-        let mut dependency_graph = compute_root_parallel_dependency_graph(parameters, has_defers)?;
+        let mut dependency_graph = compute_root_parallel_dependency_graph(
+            parameters,
+            has_defers,
+            non_local_selection_state,
+        )?;
 
         let (main, deferred) = dependency_graph.process(&mut *processor, root_kind)?;
         snapshot!(
@@ -740,13 +795,14 @@ fn compute_plan_for_defer_conditionals(
     parameters: &mut QueryPlanningParameters,
     processor: &mut FetchDependencyGraphToQueryPlanProcessor,
     defer_conditions: IndexMap<Name, IndexSet<String>>,
+    non_local_selection_state: &mut Option<non_local_selections_estimation::State>,
 ) -> Result<Option<PlanNode>, FederationError> {
     generate_condition_nodes(
         parameters.operation.clone(),
         defer_conditions.iter(),
         &mut |op| {
             parameters.operation = op;
-            compute_plan_internal(parameters, processor, true)
+            compute_plan_internal(parameters, processor, true, non_local_selection_state)
         },
     )
 }

@@ -0,0 +1,994 @@
+use apollo_compiler::Name;
+use apollo_compiler::collections::IndexMap;
+use apollo_compiler::collections::IndexSet;
+use apollo_compiler::schema::ExtendedType;
+use indexmap::map::Entry;
+use petgraph::graph::NodeIndex;
+use petgraph::visit::EdgeRef;
+use petgraph::visit::IntoNodeReferences;
+
+use crate::bail;
+use crate::error::FederationError;
+use crate::operation::Selection;
+use crate::operation::SelectionSet;
+use crate::query_graph::OverrideCondition;
+use crate::query_graph::QueryGraph;
+use crate::query_graph::QueryGraphEdgeTransition;
+use crate::query_graph::graph_path::OpPathElement;
+use crate::query_plan::query_planning_traversal::QueryPlanningTraversal;
+use crate::schema::position::CompositeTypeDefinitionPosition;
+use crate::schema::position::INTROSPECTION_TYPENAME_FIELD_NAME;
+use crate::schema::position::ObjectTypeDefinitionPosition;
+
+impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
+    pub(super) const MAX_NON_LOCAL_SELECTIONS: u64 = 100_000;
+
+    /// This calls `check_non_local_selections_limit_exceeded()` for each of the selections in the
+    /// open branches stack; see that function's doc comment for more information.
+    pub(super) fn check_non_local_selections_limit_exceeded_at_root(
+        &self,
+        state: &mut State,
+    ) -> Result<bool, FederationError> {
+        for branch in &self.open_branches {
+            let tail_nodes = branch
+                .open_branch
+                .0
+                .iter()
+                .flat_map(|option| option.paths.0.iter().map(|path| path.tail))
+                .collect::<IndexSet<_>>();
+            let tail_nodes_info = self.estimate_nodes_with_indirect_options(tail_nodes)?;
+
+            // Note that top-level selections aren't avoided via fully-local selection set
+            // optimization, so we always add them here.
+            if Self::update_count(
+                branch.selections.len(),
+                tail_nodes_info.next_nodes.len(),
+                state,
+            ) {
+                return Ok(true);
+            }
+
+            for selection in &branch.selections {
+                if let Some(selection_set) = selection.selection_set() {
+                    let selection_has_defer = selection.element()?.has_defer();
+                    let next_nodes = self.estimate_next_nodes_for_selection(
+                        &selection.element()?,
+                        &tail_nodes_info,
+                        state,
+                    )?;
+                    if self.check_non_local_selections_limit_exceeded(
+                        selection_set,
+                        &next_nodes,
+                        selection_has_defer,
+                        state,
+                    )? {
+                        return Ok(true);
+                    }
+                }
+            }
+        }
+        Ok(false)
+    }
+
+    /// When recursing through a selection set to generate options from each element, there is an
+    /// optimization that allows us to avoid option exploration if a selection set is "fully local"
+    /// from all the possible nodes we could be at in the query graph.
+    ///
+    /// This function computes an approximate upper bound on the number of selections in a selection
+    /// set that wouldn't be avoided by such an optimization (i.e. the "non-local" selections), and
+    /// adds it to the given count in the state. Note that the count for a given selection set is
+    /// scaled by an approximate upper bound on the possible number of tail nodes for paths ending
+    /// at that selection set. If at any point, the count exceeds `Self::MAX_NON_LOCAL_SELECTIONS`,
+    /// then this function will return `true`.
+    ///
+    /// This function's code is closely related to `selection_set_is_fully_local_from_all_nodes()`
+    /// (which implements the aforementioned optimization). However, when it comes to traversing the
+    /// query graph, we generally ignore the effects of edge pre-conditions and other optimizations
+    /// to option generation for efficiency's sake, giving us an upper bound since the extra nodes
+    /// may fail some of the checks (e.g. the selection set may not rebase on them).
+    ///
+    /// Note that this function takes in whether the parent selection of the selection set has
+    /// @defer, as that affects whether the optimization is disabled for that selection set.
+    fn check_non_local_selections_limit_exceeded(
+        &self,
+        selection_set: &SelectionSet,
+        parent_nodes: &NextNodesInfo,
+        parent_selection_has_defer: bool,
+        state: &mut State,
+    ) -> Result<bool, FederationError> {
+        // Compute whether the selection set is non-local, and if so, add its selections to the
+        // count. Any of the following causes the selection set to be non-local.
+        // 1. The selection set's nodes having at least one reachable cross-subgraph edge.
+        // 2. The parent selection having @defer.
+        // 3. Any selection in the selection set having @defer.
+        // 4. Any selection in the selection set being an inline fragment whose type condition
+        //    has inconsistent runtime types across subgraphs.
+        // 5. Any selection in the selection set being unable to be rebased on the selection
+        //    set's nodes.
+        // 6. Any nested selection sets causing the count to be incremented.
+        let mut selection_set_is_non_local = parent_nodes
+            .next_nodes_have_reachable_cross_subgraph_edges
+            || parent_selection_has_defer;
+        for selection in selection_set.selections.values() {
+            let element = selection.element()?;
+            let selection_has_defer = element.has_defer();
+            let selection_has_inconsistent_runtime_types =
+                if let OpPathElement::InlineFragment(inline_fragment) = element {
+                    inline_fragment
+                        .type_condition_position
+                        .map(|type_condition_pos| {
+                            self.parameters
+                                .abstract_types_with_inconsistent_runtime_types
+                                .contains(type_condition_pos.type_name())
+                        })
+                        .unwrap_or_default()
+                } else {
+                    false
+                };
+
+            let old_count = state.count;
+            if let Some(selection_set) = selection.selection_set() {
+                let next_nodes = self.estimate_next_nodes_for_selection(
+                    &selection.element()?,
+                    parent_nodes,
+                    state,
+                )?;
+                if self.check_non_local_selections_limit_exceeded(
+                    selection_set,
+                    &next_nodes,
+                    selection_has_defer,
+                    state,
+                )? {
+                    return Ok(true);
+                }
+            }
+
+            selection_set_is_non_local = selection_set_is_non_local
+                || selection_has_defer
+                || selection_has_inconsistent_runtime_types
+                || (old_count != state.count);
+        }
+        // Determine whether the selection can be rebased on all selection set nodes (without
+        // indirect options). This is more expensive, so we do this last/only if needed. Note
+        // that we were originally calling a slightly modified `can_add_to()` to mimic the logic
+        // in `selection_set_is_fully_local_from_all_nodes()`, but this ended up being rather
+        // expensive in practice, so an optimized version using precomputation is used below.
+        if !selection_set_is_non_local && !parent_nodes.next_nodes.is_empty() {
+            let metadata = self
+                .parameters
+                .federated_query_graph
+                .non_local_selection_metadata();
+            for selection in selection_set.selections.values() {
+                match selection {
+                    Selection::Field(field) => {
+                        // Note that while the precomputed metadata accounts for @fromContext,
+                        // it doesn't account for checking whether the operation field's parent
+                        // type either matches the subgraph schema's parent type name or is an
+                        // interface type. Given current composition rules, this should always
+                        // be the case when rebasing supergraph/API schema queries onto one of
+                        // its subgraph schema, so we avoid the check here for performance.
+                        let Some(rebaseable_parent_nodes) = metadata
+                            .fields_to_rebaseable_parent_nodes
+                            .get(field.field.name())
+                        else {
+                            selection_set_is_non_local = true;
+                            break;
+                        };
+                        if !parent_nodes.next_nodes.is_subset(rebaseable_parent_nodes) {
+                            selection_set_is_non_local = true;
+                            break;
+                        }
+                    }
+                    Selection::InlineFragment(inline_fragment) => {
+                        let Some(type_condition_pos) =
+                            &inline_fragment.inline_fragment.type_condition_position
+                        else {
+                            // Inline fragments without type conditions can always be rebased.
+                            continue;
+                        };
+                        let Some(rebaseable_parent_nodes) = metadata
+                            .inline_fragments_to_rebaseable_parent_nodes
+                            .get(type_condition_pos.type_name())
+                        else {
+                            selection_set_is_non_local = true;
+                            break;
+                        };
+                        if !parent_nodes.next_nodes.is_subset(rebaseable_parent_nodes) {
+                            selection_set_is_non_local = true;
+                            break;
+                        }
+                    }
+                    Selection::FragmentSpread(_) => {
+                        bail!("Unexpected fragment spread")
+                    }
+                }
+            }
+        }
+        if selection_set_is_non_local
+            && Self::update_count(
+                selection_set.selections.len(),
+                parent_nodes.next_nodes.len(),
+                state,
+            )
+        {
+            return Ok(true);
+        }
+        Ok(false)
+    }
+
+    /// Updates the non-local selection set count in the state, returning true if this causes the
+    /// count to exceed `Self::MAX_NON_LOCAL_SELECTIONS`.
+    fn update_count(num_selections: usize, num_parent_nodes: usize, state: &mut State) -> bool {
+        let Ok(num_selections) = u64::try_from(num_selections) else {
+            return true;
+        };
+        let Ok(num_parent_nodes) = u64::try_from(num_parent_nodes) else {
+            return true;
+        };
+        let Some(additional_count) = num_selections.checked_mul(num_parent_nodes) else {
+            return true;
+        };
+        if let Some(new_count) = state
+            .count
+            .checked_add(additional_count)
+            .take_if(|v| *v <= Self::MAX_NON_LOCAL_SELECTIONS)
+        {
+            state.count = new_count;
+        } else {
+            return true;
+        };
+        false
+    }
+
+    /// In `check_non_local_selections_limit_exceeded()`, when handling a given selection for a set
+    /// of parent nodes (including indirect options), this function can be used to estimate an
+    /// upper bound on the next nodes after taking the selection (also with indirect options).
+    fn estimate_next_nodes_for_selection(
+        &self,
+        element: &OpPathElement,
+        parent_nodes: &NextNodesInfo,
+        state: &mut State,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let cache = state
+            .next_nodes_cache
+            .entry(match element {
+                OpPathElement::Field(field) => {
+                    SelectionKey::Field(field.field_position.field_name().clone())
+                }
+                OpPathElement::InlineFragment(inline_fragment) => {
+                    let Some(type_condition_pos) = &inline_fragment.type_condition_position else {
+                        return Ok(parent_nodes.clone());
+                    };
+                    SelectionKey::InlineFragment(type_condition_pos.type_name().clone())
+                }
+            })
+            .or_default();
+        let mut next_nodes = NextNodesInfo::default();
+        for type_name in &parent_nodes.next_nodes_with_indirect_options.types {
+            match cache.types_to_next_nodes.entry(type_name.clone()) {
+                Entry::Occupied(entry) => next_nodes.extend(entry.get()),
+                Entry::Vacant(entry) => {
+                    let Some(indirect_options) = self
+                        .parameters
+                        .federated_query_graph
+                        .non_local_selection_metadata()
+                        .types_to_indirect_options
+                        .get(type_name)
+                    else {
+                        bail!("Unexpectedly missing node information for cached type.");
+                    };
+                    let new_next_nodes = self.estimate_next_nodes_for_selection_without_caching(
+                        element,
+                        indirect_options.same_type_options.iter(),
+                    )?;
+                    next_nodes.extend(entry.insert(new_next_nodes));
+                }
+            }
+        }
+        for node in &parent_nodes
+            .next_nodes_with_indirect_options
+            .remaining_nodes
+        {
+            match cache.remaining_nodes_to_next_nodes.entry(*node) {
+                Entry::Occupied(entry) => next_nodes.extend(entry.get()),
+                Entry::Vacant(entry) => {
+                    let new_next_nodes = self.estimate_next_nodes_for_selection_without_caching(
+                        element,
+                        std::iter::once(node),
+                    )?;
+                    next_nodes.extend(entry.insert(new_next_nodes));
+                }
+            }
+        }
+        Ok(next_nodes)
+    }
+
+    /// Estimate an upper bound on the next nodes after taking the selection on the given parent
+    /// nodes. Because we're just trying for an upper bound, we assume we can always take
+    /// type-preserving non-collecting transitions, we ignore any conditions on the selection
+    /// edge, and we always type-explode. (We do account for override conditions, which are
+    /// relatively straightforward.)
+    ///
+    /// Since we're iterating through next nodes in the process, for efficiency sake we also
+    /// compute whether there are any reachable cross-subgraph edges from the next nodes
+    /// (without indirect options). This method assumes that inline fragments have type
+    /// conditions.
+    fn estimate_next_nodes_for_selection_without_caching<'c>(
+        &self,
+        element: &OpPathElement,
+        parent_nodes: impl Iterator<Item = &'c NodeIndex>,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let mut next_nodes = IndexSet::default();
+        let nodes_to_object_type_downcasts = &self
+            .parameters
+            .federated_query_graph
+            .non_local_selection_metadata()
+            .nodes_to_object_type_downcasts;
+        match element {
+            OpPathElement::Field(field) => {
+                let Some(field_endpoints) = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .fields_to_endpoints
+                    .get(field.name())
+                else {
+                    return Ok(Default::default());
+                };
+                let mut process_head_node = |node: NodeIndex| {
+                    let Some(target) = field_endpoints.get(&node) else {
+                        return;
+                    };
+                    match target {
+                        FieldTarget::NonOverride(target) => {
+                            next_nodes.insert(*target);
+                        }
+                        FieldTarget::Override(target, condition) => {
+                            if condition.check(&self.parameters.override_conditions) {
+                                next_nodes.insert(*target);
+                            }
+                        }
+                    }
+                };
+                for node in parent_nodes {
+                    // As an upper bound for efficiency sake, we consider both non-type-exploded
+                    // and type-exploded options.
+                    process_head_node(*node);
+                    let Some(object_type_downcasts) = nodes_to_object_type_downcasts.get(node)
+                    else {
+                        continue;
+                    };
+                    match object_type_downcasts {
+                        ObjectTypeDowncasts::NonInterfaceObject(downcasts) => {
+                            for node in downcasts.values() {
+                                process_head_node(*node);
+                            }
+                        }
+                        ObjectTypeDowncasts::InterfaceObject(_) => {
+                            // Interface object fake downcasts only go back to the
+                            // self node, so we ignore them.
+                        }
+                    }
+                }
+            }
+            OpPathElement::InlineFragment(inline_fragment) => {
+                let Some(type_condition_pos) = &inline_fragment.type_condition_position else {
+                    bail!("Inline fragment unexpectedly had no type condition")
+                };
+                let inline_fragment_endpoints = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .inline_fragments_to_endpoints
+                    .get(type_condition_pos.type_name());
+                // If we end up computing runtime types for the type condition, only do it once.
+                let mut possible_runtime_types = None;
+                for node in parent_nodes {
+                    // We check whether there's already a (maybe fake) downcast edge for the
+                    // type condition (note that we've inserted fake downcasts for same-type
+                    // type conditions into the metadata).
+                    if let Some(next_node) = inline_fragment_endpoints.and_then(|e| e.get(node)) {
+                        next_nodes.insert(*next_node);
+                        continue;
+                    }
+
+                    // If not, then we need to type explode across the possible runtime types
+                    // (in the supergraph schema) for the type condition.
+                    let Some(downcasts) = nodes_to_object_type_downcasts.get(node) else {
+                        continue;
+                    };
+                    let possible_runtime_types = match &possible_runtime_types {
+                        Some(possible_runtime_types) => possible_runtime_types,
+                        None => {
+                            let type_condition_in_supergraph_pos = self
+                                .parameters
+                                .supergraph_schema
+                                .get_type(type_condition_pos.type_name().clone())?;
+                            possible_runtime_types.insert(
+                                self.parameters.supergraph_schema.possible_runtime_types(
+                                    type_condition_in_supergraph_pos.try_into()?,
+                                )?,
+                            )
+                        }
+                    };
+
+                    match downcasts {
+                        ObjectTypeDowncasts::NonInterfaceObject(downcasts) => {
+                            for (type_name, target_node) in downcasts {
+                                if possible_runtime_types.contains(&ObjectTypeDefinitionPosition {
+                                    type_name: type_name.clone(),
+                                }) {
+                                    next_nodes.insert(*target_node);
+                                }
+                            }
+                        }
+                        ObjectTypeDowncasts::InterfaceObject(downcasts) => {
+                            for type_name in downcasts {
+                                if possible_runtime_types.contains(&ObjectTypeDefinitionPosition {
+                                    type_name: type_name.clone(),
+                                }) {
+                                    // Note that interface object fake downcasts are self edges,
+                                    // so we're done once we find one.
+                                    next_nodes.insert(*node);
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        self.estimate_nodes_with_indirect_options(next_nodes)
+    }
+
+    /// Estimate the indirect options for the given next nodes, and add them to the given nodes.
+    /// As an upper bound for efficiency's sake, we assume we can take any indirect option (i.e.
+    /// ignore any edge conditions).
+    fn estimate_nodes_with_indirect_options(
+        &self,
+        next_nodes: IndexSet<NodeIndex>,
+    ) -> Result<NextNodesInfo, FederationError> {
+        let mut next_nodes_info = NextNodesInfo {
+            next_nodes,
+            ..Default::default()
+        };
+        for next_node in &next_nodes_info.next_nodes {
+            let next_node_weight = self
+                .parameters
+                .federated_query_graph
+                .node_weight(*next_node)?;
+            next_nodes_info.next_nodes_have_reachable_cross_subgraph_edges = next_nodes_info
+                .next_nodes_have_reachable_cross_subgraph_edges
+                || next_node_weight.has_reachable_cross_subgraph_edges;
+
+            let next_node_type_pos: CompositeTypeDefinitionPosition =
+                next_node_weight.type_.clone().try_into()?;
+            if let Some(options_metadata) = self
+                .parameters
+                .federated_query_graph
+                .non_local_selection_metadata()
+                .types_to_indirect_options
+                .get(next_node_type_pos.type_name())
+            {
+                // If there's an entry in `types_to_indirect_options` for the type, then the
+                // complete digraph for T is non-empty, so we add its type. If it's our first
+                // time seeing this type, we also add any of the complete digraph's interface
+                // object options.
+                if next_nodes_info
+                    .next_nodes_with_indirect_options
+                    .types
+                    .insert(next_node_type_pos.type_name().clone())
+                {
+                    next_nodes_info
+                        .next_nodes_with_indirect_options
+                        .types
+                        .extend(options_metadata.interface_object_options.iter().cloned());
+                }
+                // If the node is a member of the complete digraph, then we don't need to
+                // separately add the remaining node.
+                if options_metadata.same_type_options.contains(next_node) {
+                    continue;
+                }
+            }
+            // We need to add the remaining node, and if its our first time seeing it, we also
+            // add any of its interface object options.
+            if next_nodes_info
+                .next_nodes_with_indirect_options
+                .remaining_nodes
+                .insert(*next_node)
+            {
+                if let Some(options) = self
+                    .parameters
+                    .federated_query_graph
+                    .non_local_selection_metadata()
+                    .remaining_nodes_to_interface_object_options
+                    .get(next_node)
+                {
+                    next_nodes_info
+                        .next_nodes_with_indirect_options
+                        .types
+                        .extend(options.iter().cloned());
+                }
+            }
+        }
+
+        Ok(next_nodes_info)
+    }
+}
+
+/// Precompute relevant metadata about the query graph for speeding up the estimation of the
+/// count of non-local selections. Note that none of the algorithms used in this function should
+/// take any longer algorithmically as the rest of query graph creation (and similarly for
+/// query graph memory).
+pub(crate) fn precompute_non_local_selection_metadata(
+    graph: &QueryGraph,
+) -> Result<QueryGraphMetadata, FederationError> {
+    let mut nodes_to_interface_object_options: IndexMap<NodeIndex, IndexSet<Name>> =
+        Default::default();
+    let mut metadata = QueryGraphMetadata::default();
+
+    for edge_ref in graph.graph().edge_references() {
+        match &edge_ref.weight().transition {
+            QueryGraphEdgeTransition::FieldCollection {
+                field_definition_position,
+                ..
+            } => {
+                // We skip selections where the tail is a non-composite type, as we'll never
+                // need to estimate the next nodes for such selections.
+                let Ok(_): Result<CompositeTypeDefinitionPosition, _> = graph
+                    .node_weight(edge_ref.target())?
+                    .type_
+                    .clone()
+                    .try_into()
+                else {
+                    continue;
+                };
+                let target = edge_ref
+                    .weight()
+                    .override_condition
+                    .clone()
+                    .map(|condition| FieldTarget::Override(edge_ref.target(), condition))
+                    .unwrap_or_else(|| FieldTarget::NonOverride(edge_ref.target()));
+                metadata
+                    .fields_to_endpoints
+                    .entry(field_definition_position.field_name().clone())
+                    .or_default()
+                    .insert(edge_ref.source(), target);
+            }
+            QueryGraphEdgeTransition::Downcast {
+                to_type_position, ..
+            } => {
+                if to_type_position.is_object_type() {
+                    let ObjectTypeDowncasts::NonInterfaceObject(downcasts) = metadata
+                        .nodes_to_object_type_downcasts
+                        .entry(edge_ref.source())
+                        .or_insert_with(|| {
+                            ObjectTypeDowncasts::NonInterfaceObject(Default::default())
+                        })
+                    else {
+                        bail!("Unexpectedly found interface object with regular object downcasts")
+                    };
+                    downcasts.insert(to_type_position.type_name().clone(), edge_ref.target());
+                }
+                metadata
+                    .inline_fragments_to_endpoints
+                    .entry(to_type_position.type_name().clone())
+                    .or_default()
+                    .insert(edge_ref.source(), edge_ref.target());
+            }
+            QueryGraphEdgeTransition::InterfaceObjectFakeDownCast { to_type_name, .. } => {
+                // Note that fake downcasts for interface objects are only created to "fake"
+                // object types.
+                let ObjectTypeDowncasts::InterfaceObject(downcasts) = metadata
+                    .nodes_to_object_type_downcasts
+                    .entry(edge_ref.source())
+                    .or_insert_with(|| ObjectTypeDowncasts::InterfaceObject(Default::default()))
+                else {
+                    bail!("Unexpectedly found abstract type with interface object downcasts")
+                };
+                downcasts.insert(to_type_name.clone());
+                metadata
+                    .inline_fragments_to_endpoints
+                    .entry(to_type_name.clone())
+                    .or_default()
+                    .insert(edge_ref.source(), edge_ref.target());
+            }
+            QueryGraphEdgeTransition::KeyResolution
+            | QueryGraphEdgeTransition::RootTypeResolution { .. } => {
+                let head_type_pos: CompositeTypeDefinitionPosition = graph
+                    .node_weight(edge_ref.source())?
+                    .type_
+                    .clone()
+                    .try_into()?;
+                let tail_type_pos: CompositeTypeDefinitionPosition = graph
+                    .node_weight(edge_ref.target())?
+                    .type_
+                    .clone()
+                    .try_into()?;
+                if head_type_pos.type_name() == tail_type_pos.type_name() {
+                    // In this case, we have a non-interface-object key resolution edge or a
+                    // root type resolution edge. The tail must be part of the complete digraph
+                    // for the tail's type, so we record the member.
+                    metadata
+                        .types_to_indirect_options
+                        .entry(tail_type_pos.type_name().clone())
+                        .or_default()
+                        .same_type_options
+                        .insert(edge_ref.target());
+                } else {
+                    // Otherwise, this must be an interface object key resolution edge. We don't
+                    // know the members of the complete digraph for the head's type yet, so we
+                    // can't set the metadata yet, and instead store the head to interface
+                    // object type mapping in a temporary map.
+                    nodes_to_interface_object_options
+                        .entry(edge_ref.source())
+                        .or_default()
+                        .insert(tail_type_pos.type_name().clone());
+                }
+            }
+            QueryGraphEdgeTransition::SubgraphEnteringTransition => {}
+        }
+    }
+
+    // Now that we've finished computing members of the complete digraphs, we can properly track
+    // interface object options.
+    for (node, options) in nodes_to_interface_object_options {
+        let node_type_pos: CompositeTypeDefinitionPosition =
+            graph.node_weight(node)?.type_.clone().try_into()?;
+        if let Some(options_metadata) = metadata
+            .types_to_indirect_options
+            .get_mut(node_type_pos.type_name())
+        {
+            if options_metadata.same_type_options.contains(&node) {
+                options_metadata
+                    .interface_object_options
+                    .extend(options.into_iter());
+                continue;
+            }
+        }
+        metadata
+            .remaining_nodes_to_interface_object_options
+            .insert(node, options);
+    }
+
+    // The interface object options for the complete digraphs are now correct, but we need to
+    // subtract these from any interface object options for remaining nodes.
+    for (node, options) in metadata
+        .remaining_nodes_to_interface_object_options
+        .iter_mut()
+    {
+        let node_type_pos: CompositeTypeDefinitionPosition =
+            graph.node_weight(*node)?.type_.clone().try_into()?;
+        let Some(IndirectOptionsMetadata {
+            interface_object_options,
+            ..
+        }) = metadata
+            .types_to_indirect_options
+            .get(node_type_pos.type_name())
+        else {
+            continue;
+        };
+        options.retain(|type_name| !interface_object_options.contains(type_name));
+    }
+
+    // If this subtraction left any interface object option sets empty, we remove them.
+    metadata
+        .remaining_nodes_to_interface_object_options
+        .retain(|_, options| !options.is_empty());
+
+    // For all composite type nodes, we pretend that there's a self-downcast edge for that type,
+    // as this simplifies next node calculation.
+    for (node, node_weight) in graph.graph().node_references() {
+        let Ok(node_type_pos): Result<CompositeTypeDefinitionPosition, _> =
+            node_weight.type_.clone().try_into()
+        else {
+            continue;
+        };
+        metadata
+            .inline_fragments_to_endpoints
+            .entry(node_type_pos.type_name().clone())
+            .or_default()
+            .insert(node, node);
+        if node_type_pos.is_object_type()
+            && !graph
+                .schema_by_source(&node_weight.source)?
+                .is_interface_object_type(node_type_pos.clone().into())?
+        {
+            let ObjectTypeDowncasts::NonInterfaceObject(downcasts) = metadata
+                .nodes_to_object_type_downcasts
+                .entry(node)
+                .or_insert_with(|| ObjectTypeDowncasts::NonInterfaceObject(Default::default()))
+            else {
+                bail!(
+                    "Unexpectedly found object type with interface object downcasts in supergraph"
+                )
+            };
+            downcasts.insert(node_type_pos.type_name().clone(), node);
+        }
+    }
+
+    // For each subgraph schema, we iterate through its composite types, so that we can collect
+    // metadata relevant to rebasing.
+    for (subgraph_name, subgraph_schema) in graph.subgraph_schemas() {
+        // We pass through each composite type, recording whether the field can be rebased on it
+        // along with interface implements/union membership relationships.
+        let mut fields_to_rebaseable_types: IndexMap<Name, IndexSet<Name>> = Default::default();
+        let mut object_types_to_implementing_composite_types: IndexMap<Name, IndexSet<Name>> =
+            Default::default();
+        let Some(subgraph_metadata) = subgraph_schema.subgraph_metadata() else {
+            bail!("Subgraph schema unexpectedly did not have subgraph metadata")
+        };
+        let from_context_directive_definition_name = &subgraph_metadata
+            .federation_spec_definition()
+            .from_context_directive_definition(subgraph_schema)?
+            .name;
+        for (type_name, type_) in &subgraph_schema.schema().types {
+            match type_ {
+                ExtendedType::Object(type_) => {
+                    // Record fields that don't contain @fromContext as being rebaseable (also
+                    // including __typename).
+                    for (field_name, field_definition) in &type_.fields {
+                        if field_definition.arguments.iter().any(|arg_definition| {
+                            arg_definition
+                                .directives
+                                .has(from_context_directive_definition_name)
+                        }) {
+                            continue;
+                        }
+                        fields_to_rebaseable_types
+                            .entry(field_name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                    // Record the object type as implementing itself.
+                    let implementing_composite_types = object_types_to_implementing_composite_types
+                        .entry(type_name.clone())
+                        .or_default();
+                    implementing_composite_types.insert(type_name.clone());
+                    // For each implements, record the interface type as an implementing type.
+                    if !type_.implements_interfaces.is_empty() {
+                        implementing_composite_types.extend(
+                            type_
+                                .implements_interfaces
+                                .iter()
+                                .map(|interface_name| interface_name.name.clone()),
+                        );
+                    }
+                }
+                ExtendedType::Interface(type_) => {
+                    // Record fields that don't contain @fromContext as being rebaseable (also
+                    // including __typename).
+                    for (field_name, field_definition) in &type_.fields {
+                        if field_definition.arguments.iter().any(|arg_definition| {
+                            arg_definition
+                                .directives
+                                .has(from_context_directive_definition_name)
+                        }) {
+                            continue;
+                        }
+                        fields_to_rebaseable_types
+                            .entry(field_name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                }
+                ExtendedType::Union(type_) => {
+                    // Just record the __typename field as being rebaseable.
+                    fields_to_rebaseable_types
+                        .entry(INTROSPECTION_TYPENAME_FIELD_NAME.clone())
+                        .or_default()
+                        .insert(type_name.clone());
+                    // For each member, record the union type as an implementing type.
+                    for member_name in &type_.members {
+                        object_types_to_implementing_composite_types
+                            .entry(member_name.name.clone())
+                            .or_default()
+                            .insert(type_name.clone());
+                    }
+                }
+                _ => {}
+            }
+        }
+
+        // With the interface implements/union membership relationships, we can compute which
+        // pairs of types have at least one possible runtime type in their intersection, and
+        // are thus rebaseable.
+        let mut inline_fragments_to_rebaseable_types: IndexMap<Name, IndexSet<Name>> =
+            Default::default();
+        for implementing_types in object_types_to_implementing_composite_types.values() {
+            for type_name in implementing_types {
+                inline_fragments_to_rebaseable_types
+                    .entry(type_name.clone())
+                    .or_default()
+                    .extend(implementing_types.iter().cloned())
+            }
+        }
+
+        // Finally, we can compute the nodes for the rebaseable types, as we'll be working with
+        // those instead of types when checking whether an operation element can be rebased.
+        let types_to_nodes = graph.types_to_nodes_by_source(subgraph_name)?;
+        for (field_name, types) in fields_to_rebaseable_types {
+            metadata
+                .fields_to_rebaseable_parent_nodes
+                .entry(field_name)
+                .or_default()
+                .extend(
+                    types
+                        .iter()
+                        .flat_map(|type_| types_to_nodes.get(type_).map(|nodes| nodes.iter()))
+                        .flatten()
+                        .cloned(),
+                );
+        }
+        for (type_condition_name, types) in inline_fragments_to_rebaseable_types {
+            metadata
+                .inline_fragments_to_rebaseable_parent_nodes
+                .entry(type_condition_name)
+                .or_default()
+                .extend(
+                    types
+                        .iter()
+                        .flat_map(|type_| types_to_nodes.get(type_).map(|nodes| nodes.iter()))
+                        .flatten()
+                        .cloned(),
+                )
+        }
+    }
+    Ok(metadata)
+}
+
+/// During query graph creation, we pre-compute metadata that helps us greatly speed up the
+/// estimation process during request execution. The expected time and memory consumed for
+/// pre-computation is (in the worst case) expected to be on the order of the number of nodes
+/// plus the number of edges.
+///
+/// Note that when the below field docs talk about a "complete digraph", they are referring to
+/// the graph theory concept: https://en.wikipedia.org/wiki/Complete_graph
+#[derive(Debug, Default)]
+pub(crate) struct QueryGraphMetadata {
+    /// When a (resolvable) @key exists on a type T in a subgraph, a key resolution edge is
+    /// created from every subgraph's type T to that subgraph's type T. This similarly holds for
+    /// root type resolution edges. This means that the nodes of type T with such a @key (or are
+    /// operation root types) form a complete digraph in the query graph. These indirect options
+    /// effectively occur as a group in our estimation process, so we track group members here
+    /// per type name, and precompute units of work relative to these groups.
+    ///
+    /// Interface object types I in a subgraph will only sometimes create a key resolution edge
+    /// between an implementing type T in a subgraph and that subgraph's type I. This means the
+    /// nodes of the complete digraph for I are indirect options for such nodes of type T. We
+    /// track any such types I that are reachable for at least one node in the complete digraph
+    /// for type T here as well.
+    types_to_indirect_options: IndexMap<Name, IndirectOptionsMetadata>,
+    /// For nodes of a type T that aren't in their complete digraph (due to not having a @key),
+    /// these remaining nodes will have the complete digraph of T (and any interface object
+    /// complete digraphs) as indirect options, but these remaining nodes may separately have
+    /// more indirect options that are not options for the complete digraph of T, specifically
+    /// if the complete digraph for T has no key resolution edges to an interface object I, but
+    /// this remaining node does. We keep track of such interface object types for those
+    /// remaining nodes here.
+    remaining_nodes_to_interface_object_options: IndexMap<NodeIndex, IndexSet<Name>>,
+    /// A map of field names to the endpoints of field query graph edges with that field name. Note
+    /// we additionally store the progressive overrides label, if the edge is conditioned on it.
+    fields_to_endpoints: IndexMap<Name, IndexMap<NodeIndex, FieldTarget>>,
+    /// A map of type condition names to endpoints of downcast query graph edges with that type
+    /// condition name, including fake downcasts for interface objects, and a non-existent edge that
+    /// represents a type condition name equal to the parent type.
+    inline_fragments_to_endpoints: IndexMap<Name, IndexMap<NodeIndex, NodeIndex>>,
+    /// A map of composite type nodes to their downcast edges that lead specifically to an object
+    /// type (i.e., the possible runtime types of the node's type).
+    nodes_to_object_type_downcasts: IndexMap<NodeIndex, ObjectTypeDowncasts>,
+    /// A map of field names to parent nodes whose corresponding type and schema can be rebased on
+    /// by the field.
+    fields_to_rebaseable_parent_nodes: IndexMap<Name, IndexSet<NodeIndex>>,
+    /// A map of type condition names to parent nodes whose corresponding type and schema can be
+    /// rebased on by an inline fragment with that type condition.
+    inline_fragments_to_rebaseable_parent_nodes: IndexMap<Name, IndexSet<NodeIndex>>,
+}
+
+/// Indirect option metadata for the complete digraph for type T. See [QueryGraphMetadata] for
+/// more information about how we group indirect options into complete digraphs.
+#[derive(Debug, Default)]
+pub(crate) struct IndirectOptionsMetadata {
+    /// The members of the complete digraph for type T.
+    same_type_options: IndexSet<NodeIndex>,
+    /// Any interface object types I that are reachable for at least one node in the complete
+    /// digraph for type T.
+    interface_object_options: IndexSet<Name>,
+}
+
+#[derive(Debug)]
+enum FieldTarget {
+    /// Normal non-overridden fields, which don't have a label condition.
+    NonOverride(NodeIndex),
+    /// Overridden fields, which have a label condition.
+    Override(NodeIndex, OverrideCondition),
+}
+
+#[derive(Debug)]
+enum ObjectTypeDowncasts {
+    /// Normal non-interface-object types have regular downcasts to their object type nodes.
+    NonInterfaceObject(IndexMap<Name, NodeIndex>),
+    /// Interface object types have "fake" downcasts to nodes that are really the self node.
+    InterfaceObject(IndexSet<Name>),
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct State {
+    /// An estimation of the number of non-local selections for the whole operation (where the count
+    /// for a given selection set is scaled by the number of tail nodes at that selection set). Note
+    /// this does not count selections from recursive query planning.
+    pub(crate) count: u64,
+    /// Whenever we take a selection on a set of nodes with indirect options, we cache the
+    /// resulting nodes here.
+    next_nodes_cache: IndexMap<SelectionKey, NextNodesCache>,
+}
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+enum SelectionKey {
+    /// For field selections, this is the field's name.
+    Field(Name),
+    /// For inline fragment selections, this is the type condition's name.
+    InlineFragment(Name),
+}
+
+/// See [QueryGraphMetadata] for more information about how we group indirect options into
+/// complete digraphs.
+#[derive(Debug, Default)]
+struct NextNodesCache {
+    /// This is the merged next node info for selections on the set of nodes in the complete
+    /// digraph for the given type T. Note that this does not merge in the next node info for
+    /// any interface object options reachable from nodes in that complete digraph for T.
+    types_to_next_nodes: IndexMap<Name, NextNodesInfo>,
+    /// This is the next node info for selections on the given node. Note that this does not
+    /// merge in the next node info for any interface object options reachable from that node.
+    remaining_nodes_to_next_nodes: IndexMap<NodeIndex, NextNodesInfo>,
+}
+
+#[derive(Clone, Debug, Default)]
+struct NextNodesInfo {
+    /// The next nodes after taking the selection.
+    next_nodes: IndexSet<NodeIndex>,
+    /// Whether any cross-subgraph edges are reachable from any next nodes.
+    next_nodes_have_reachable_cross_subgraph_edges: bool,
+    /// These are the next nodes along with indirect options, represented succinctly by the
+    /// types of any complete digraphs along with remaining nodes.
+    next_nodes_with_indirect_options: NodesWithIndirectOptionsInfo,
+}
+
+impl NextNodesInfo {
+    fn extend(&mut self, other: &Self) {
+        self.next_nodes.extend(other.next_nodes.iter().cloned());
+        self.next_nodes_have_reachable_cross_subgraph_edges = self
+            .next_nodes_have_reachable_cross_subgraph_edges
+            || other.next_nodes_have_reachable_cross_subgraph_edges;
+        self.next_nodes_with_indirect_options
+            .extend(&other.next_nodes_with_indirect_options);
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+struct NodesWithIndirectOptionsInfo {
+    /// For indirect options that are representable as complete digraphs for a type T, these are
+    /// those types.
+    types: IndexSet<Name>,
+    /// For any nodes of type T that aren't in their complete digraphs for type T, these are
+    /// those nodes.
+    remaining_nodes: IndexSet<NodeIndex>,
+}
+
+impl NodesWithIndirectOptionsInfo {
+    fn extend(&mut self, other: &Self) {
+        self.types.extend(other.types.iter().cloned());
+        self.remaining_nodes
+            .extend(other.remaining_nodes.iter().cloned());
+    }
+}

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use apollo_compiler::Name;
 use apollo_compiler::collections::IndexSet;
 use petgraph::graph::EdgeIndex;
 use petgraph::graph::NodeIndex;
@@ -44,13 +45,14 @@ use crate::query_plan::query_planner::QueryPlannerConfig;
 use crate::query_plan::query_planner::QueryPlanningStatistics;
 use crate::query_plan::query_planner::compute_root_fetch_groups;
 use crate::schema::ValidFederationSchema;
-use crate::schema::position::AbstractTypeDefinitionPosition;
 use crate::schema::position::CompositeTypeDefinitionPosition;
 use crate::schema::position::ObjectTypeDefinitionPosition;
 use crate::schema::position::SchemaRootDefinitionKind;
 use crate::utils::logging::format_open_branch;
 use crate::utils::logging::snapshot;
 
+pub(crate) mod non_local_selections_estimation;
+
 #[cfg(feature = "snapshot_tracing")]
 mod snapshot_helper {
     // A module to import functions only used within `snapshot!(...)` macros.
@@ -81,8 +83,7 @@ pub(crate) struct QueryPlanningParameters<'a> {
     /// subgraphs.
     // PORT_NOTE: Named `inconsistentAbstractTypesRuntimes` in the JS codebase, which was slightly
     // confusing.
-    pub(crate) abstract_types_with_inconsistent_runtime_types:
-        Arc<IndexSet<AbstractTypeDefinitionPosition>>,
+    pub(crate) abstract_types_with_inconsistent_runtime_types: Arc<IndexSet<Name>>,
     /// The configuration for the query planner.
     pub(crate) config: QueryPlannerConfig,
     pub(crate) statistics: &'a QueryPlanningStatistics,
@@ -227,6 +228,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
         has_defers: bool,
         root_kind: SchemaRootDefinitionKind,
         cost_processor: FetchDependencyGraphToCostProcessor,
+        non_local_selection_state: Option<&mut non_local_selections_estimation::State>,
     ) -> Result<Self, FederationError> {
         Self::new_inner(
             parameters,
@@ -235,6 +237,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
             parameters.fetch_id_generator.clone(),
             root_kind,
             cost_processor,
+            non_local_selection_state,
             Default::default(),
             Default::default(),
             Default::default(),
@@ -254,6 +257,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
         id_generator: Arc<FetchIdGenerator>,
         root_kind: SchemaRootDefinitionKind,
         cost_processor: FetchDependencyGraphToCostProcessor,
+        non_local_selection_state: Option<&mut non_local_selections_estimation::State>,
         initial_context: OpGraphPathContext,
         excluded_destinations: ExcludedDestinations,
         excluded_conditions: ExcludedConditions,
@@ -311,6 +315,20 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
 
         traversal.open_branches = map_options_to_selections(selection_set, initial_options);
 
+        if let Some(non_local_selection_state) = non_local_selection_state {
+            if traversal
+                .check_non_local_selections_limit_exceeded_at_root(non_local_selection_state)?
+            {
+                return Err(SingleFederationError::QueryPlanComplexityExceeded {
+                    message: format!(
+                        "Number of non-local selections exceeds limit of {}",
+                        Self::MAX_NON_LOCAL_SELECTIONS,
+                    ),
+                }
+                .into());
+            }
+        }
+
         Ok(traversal)
     }
 
@@ -615,8 +633,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
                             Some(type_condition) => Ok(self
                                 .parameters
                                 .abstract_types_with_inconsistent_runtime_types
-                                .iter()
-                                .any(|ty| ty.type_name() == type_condition.type_name())),
+                                .contains(type_condition.type_name())),
                             None => Ok(false),
                         }
                     }
@@ -1138,6 +1155,7 @@ impl<'a: 'b, 'b> QueryPlanningTraversal<'a, 'b> {
             self.id_generator.clone(),
             self.root_kind,
             self.cost_processor,
+            None,
             context.clone(),
             excluded_destinations.clone(),
             excluded_conditions.add_item(edge_conditions),

@@ -24,7 +24,8 @@ fn it_handles_progressive_override_on_root_fields() {
           }
         "#,
         QueryPlanOptions {
-            override_conditions: vec!["test".to_string()]
+            override_conditions: vec!["test".to_string()],
+            ..Default::default()
         },
         @r###"
         QueryPlan {
@@ -117,7 +118,8 @@ fn it_handles_progressive_override_on_entity_fields() {
           }
         "#,
         QueryPlanOptions {
-            override_conditions: vec!["test".to_string()]
+            override_conditions: vec!["test".to_string()],
+            ..Default::default()
         },
         @r###"
           QueryPlan {
@@ -276,7 +278,8 @@ fn it_handles_progressive_override_on_nested_entity_fields() {
           }
         "#,
         QueryPlanOptions {
-            override_conditions: vec!["test".to_string()]
+            override_conditions: vec!["test".to_string()],
+            ..Default::default()
         },
         @r###"
           QueryPlan {

@@ -45,7 +45,8 @@ fn it_overrides_to_s2_when_label_is_provided() {
           }
         "#,
         QueryPlanOptions {
-            override_conditions: vec!["test".to_string()]
+            override_conditions: vec!["test".to_string()],
+            ..Default::default()
         },
         @r###"
           QueryPlan {
@@ -157,7 +158,8 @@ fn it_overrides_f1_to_s3_when_label_is_provided() {
           }
         "#,
         QueryPlanOptions {
-            override_conditions: vec!["test".to_string()]
+            override_conditions: vec!["test".to_string()],
+            ..Default::default()
         },
         @r###"
           QueryPlan {

@@ -454,6 +454,15 @@ impl InstrumentData {
             "opt.apollo.dev".to_string(),
             env_var_exists("APOLLO_ROUTER_DEV_ENV"),
         );
+        attributes.insert(
+            "opt.security.recursive_selections".to_string(),
+            crate::services::layers::query_analysis::recursive_selections_check_enabled().into(),
+        );
+        attributes.insert(
+            "opt.security.non_local_selections".to_string(),
+            crate::query_planner::query_planner_service::non_local_selections_check_enabled()
+                .into(),
+        );
 
         self.data
             .insert("apollo.router.config.env".to_string(), (1, attributes));

@@ -1,6 +1,7 @@
 ---
 source: apollo-router/src/configuration/metrics.rs
-expression: "&metrics.non_zero()"
+expression: "& metrics.non_zero()"
+snapshot_kind: text
 ---
 - name: apollo.router.config.env
   data:
@@ -14,4 +15,5 @@ expression: "&metrics.non_zero()"
           opt.apollo.license.path: true
           opt.apollo.supergraph.path: true
           opt.apollo.supergraph.urls: true
-
+          opt.security.non_local_selections: true
+          opt.security.recursive_selections: true

@@ -4,6 +4,7 @@ use std::fmt::Debug;
 use std::ops::ControlFlow;
 use std::sync::Arc;
 use std::sync::Mutex;
+use std::sync::OnceLock;
 use std::task::Poll;
 use std::time::Instant;
 
@@ -58,6 +59,25 @@ pub(crate) const RUST_QP_MODE: &str = "rust";
 const UNSUPPORTED_FED1: &str = "fed1";
 const INTERNAL_INIT_ERROR: &str = "internal";
 
+const ENV_DISABLE_NON_LOCAL_SELECTIONS_CHECK: &str =
+    "APOLLO_ROUTER_DISABLE_SECURITY_NON_LOCAL_SELECTIONS_CHECK";
+/// Should we enforce the non-local selections limit? Default true, can be toggled off with an
+/// environment variable.
+///
+/// Disabling this check is very much not advisable and we don't expect that anyone will need to do
+/// it. In the extremely unlikely case that the new protection breaks someone's legitimate queries,
+/// though, they could temporarily disable this individual limit so they can still benefit from the
+/// other new limits, until we improve the detection.
+pub(crate) fn non_local_selections_check_enabled() -> bool {
+    static ON: OnceLock<bool> = OnceLock::new();
+    *ON.get_or_init(|| {
+        let disabled =
+            std::env::var(ENV_DISABLE_NON_LOCAL_SELECTIONS_CHECK).as_deref() == Ok("true");
+
+        !disabled
+    })
+}
+
 /// A query planner that calls out to the apollo-federation crate.
 ///
 /// No caching is performed. To cache, wrap in a [`CachingQueryPlanner`].
@@ -136,6 +156,7 @@ impl QueryPlannerService {
 
             let query_plan_options = QueryPlanOptions {
                 override_conditions: plan_options.override_conditions,
+                non_local_selections_limit_enabled: non_local_selections_check_enabled(),
             };
 
             let result = operation

@@ -3,11 +3,16 @@ use std::fmt::Display;
 use std::fmt::Formatter;
 use std::hash::Hash;
 use std::sync::Arc;
+use std::sync::OnceLock;
 
 use apollo_compiler::ExecutableDocument;
+use apollo_compiler::Name;
 use apollo_compiler::Node;
 use apollo_compiler::ast;
 use apollo_compiler::executable::Operation;
+use apollo_compiler::executable::Selection;
+use apollo_compiler::executable::SelectionSet;
+use apollo_compiler::response::GraphQLError;
 use apollo_compiler::validation::Valid;
 use http::StatusCode;
 use lru::LruCache;
@@ -21,6 +26,7 @@ use crate::apollo_studio_interop::generate_extended_references;
 use crate::compute_job;
 use crate::context::OPERATION_KIND;
 use crate::context::OPERATION_NAME;
+use crate::error::ValidationErrors;
 use crate::graphql::Error;
 use crate::graphql::ErrorExtension;
 use crate::graphql::IntoGraphQLErrors;
@@ -36,6 +42,25 @@ use crate::spec::QueryHash;
 use crate::spec::Schema;
 use crate::spec::SpecError;
 
+const ENV_DISABLE_RECURSIVE_SELECTIONS_CHECK: &str =
+    "APOLLO_ROUTER_DISABLE_SECURITY_RECURSIVE_SELECTIONS_CHECK";
+/// Should we enforce the recursive selections limit? Default true, can be toggled off with an
+/// environment variable.
+///
+/// Disabling this check is very much not advisable and we don't expect that anyone will need to do
+/// it. In the extremely unlikely case that the new protection breaks someone's legitimate queries,
+/// though, they could temporarily disable this individual limit so they can still benefit from the
+/// other new limits, until we improve the detection.
+pub(crate) fn recursive_selections_check_enabled() -> bool {
+    static ON: OnceLock<bool> = OnceLock::new();
+    *ON.get_or_init(|| {
+        let disabled =
+            std::env::var(ENV_DISABLE_RECURSIVE_SELECTIONS_CHECK).as_deref() == Ok("true");
+
+        !disabled
+    })
+}
+
 /// [`Layer`] for QueryAnalysis implementation.
 #[derive(Clone)]
 #[allow(clippy::type_complexity)]
@@ -54,6 +79,8 @@ struct QueryAnalysisKey {
 }
 
 impl QueryAnalysisLayer {
+    const MAX_RECURSIVE_SELECTIONS: u32 = 10_000_000;
+
     pub(crate) async fn new(schema: Arc<Schema>, configuration: Arc<Configuration>) -> Self {
         let enable_authorization_directives =
             AuthorizationPlugin::enable_directives(&configuration, &schema).unwrap_or(false);
@@ -98,6 +125,34 @@ impl QueryAnalysisLayer {
                     schema.as_ref(),
                     conf.as_ref(),
                 )
+                .and_then(|doc| {
+                    let recursive_selections = Self::count_recursive_selections(
+                        &doc.executable,
+                        &mut Default::default(),
+                        &doc.operation.selection_set,
+                        0,
+                    );
+                    if recursive_selections.is_none() {
+                        if recursive_selections_check_enabled() {
+                            return Err(SpecError::ValidationError(ValidationErrors {
+                                errors: vec![GraphQLError {
+                                    message:
+                                        "Maximum recursive selections limit exceeded in this operation"
+                                            .to_string(),
+                                    locations: Default::default(),
+                                    path: Default::default(),
+                                    extensions: Default::default(),
+                                }],
+                            }))
+                        }
+                        tracing::info!(
+                            operation_name = ?operation_name,
+                            limit = Self::MAX_RECURSIVE_SELECTIONS,
+                            "operation exceeded maximum recursive selections limit, but limit is forcefully disabled",
+                        );
+                    }
+                    Ok(doc)
+                })
             })
         };
         // TODO: is this correct?
@@ -107,6 +162,66 @@ impl QueryAnalysisLayer {
             .expect("Query::parse_document panicked")
     }
 
+    /// Measure the number of selections that would be encountered if we walked the given selection
+    /// set while recursing into fragment spreads, and add it to the given count. `None` is returned
+    /// instead if this number exceeds `Self::MAX_RECURSIVE_SELECTIONS`.
+    ///
+    /// This function assumes that fragments referenced by spreads exist and that they don't form
+    /// cycles. If a fragment spread appears multiple times for the same named fragment, it is
+    /// counted multiple times.
+    fn count_recursive_selections<'a>(
+        document: &'a Valid<ExecutableDocument>,
+        fragment_cache: &mut HashMap<&'a Name, u32>,
+        selection_set: &'a SelectionSet,
+        mut count: u32,
+    ) -> Option<u32> {
+        for selection in &selection_set.selections {
+            count = count
+                .checked_add(1)
+                .take_if(|v| *v <= Self::MAX_RECURSIVE_SELECTIONS)?;
+            match selection {
+                Selection::Field(field) => {
+                    count = Self::count_recursive_selections(
+                        document,
+                        fragment_cache,
+                        &field.selection_set,
+                        count,
+                    )?;
+                }
+                Selection::InlineFragment(fragment) => {
+                    count = Self::count_recursive_selections(
+                        document,
+                        fragment_cache,
+                        &fragment.selection_set,
+                        count,
+                    )?;
+                }
+                Selection::FragmentSpread(fragment) => {
+                    let name = &fragment.fragment_name;
+                    if let Some(cached) = fragment_cache.get(name) {
+                        count = count
+                            .checked_add(*cached)
+                            .take_if(|v| *v <= Self::MAX_RECURSIVE_SELECTIONS)?;
+                    } else {
+                        let old_count = count;
+                        count = Self::count_recursive_selections(
+                            document,
+                            fragment_cache,
+                            &document
+                                .fragments
+                                .get(&fragment.fragment_name)
+                                .expect("validation should have ensured referenced fragments exist")
+                                .selection_set,
+                            count,
+                        )?;
+                        fragment_cache.insert(name, count - old_count);
+                    };
+                }
+            }
+        }
+        Some(count)
+    }
+
     pub(crate) async fn supergraph_request(
         &self,
         request: SupergraphRequest,

@@ -139,29 +139,29 @@ fn count<'a>(
         match selection {
             executable::Selection::Field(field) => {
                 let nested = count(document, fragment_cache, &field.selection_set);
-                counts.depth = counts.depth.max(1 + nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.depth = counts.depth.max(nested.depth.saturating_add(1));
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
                 // Multiple aliases for the same field could use different arguments
                 // Until we do full merging for limit checking purpose,
                 // approximate measured height with an upper bound rather than a lower bound.
                 let used_name = if let Some(alias) = &field.alias {
-                    counts.aliases += 1;
+                    counts.aliases = counts.aliases.saturating_add(1);
                     alias
                 } else {
                     &field.name
                 };
                 let not_seen_before = fields_seen.insert(used_name);
                 if not_seen_before {
-                    counts.height += 1;
-                    counts.root_fields += 1;
+                    counts.height = counts.height.saturating_add(1);
+                    counts.root_fields = counts.root_fields.saturating_add(1);
                 }
             }
             executable::Selection::InlineFragment(fragment) => {
                 let nested = count(document, fragment_cache, &fragment.selection_set);
                 counts.depth = counts.depth.max(nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
             }
             executable::Selection::FragmentSpread(fragment) => {
                 let name = &fragment.fragment_name;
@@ -190,8 +190,8 @@ fn count<'a>(
                     Some(Computation::Done(cached)) => nested = *cached,
                 }
                 counts.depth = counts.depth.max(nested.depth);
-                counts.height += nested.height;
-                counts.aliases += nested.aliases;
+                counts.height = counts.height.saturating_add(nested.height);
+                counts.aliases = counts.aliases.saturating_add(nested.aliases);
             }
         }
     }

@@ -0,0 +1,7 @@
+### Certain query patterns may cause resource exhaustion
+
+Corrects a set of denial-of-service (DOS) vulnerabilities that made it possible for an attacker to render router inoperable with certain simple query patterns due to uncontrolled resource consumption. All prior-released versions and configurations are vulnerable except those where `persisted_queries.enabled`, `persisted_queries.safelist.enabled`, and `persisted_queries.safelist.require_id` are all `true`.
+
+See the associated GitHub Advisories [GHSA-3j43-9v8v-cp3f](https://github.com/apollographql/router/security/advisories/GHSA-3j43-9v8v-cp3f), [GHSA-84m6-5m72-45fp](https://github.com/apollographql/router/security/advisories/GHSA-84m6-5m72-45fp), [GHSA-75m2-jhh5-j5g2](https://github.com/apollographql/router/security/advisories/GHSA-75m2-jhh5-j5g2), and [GHSA-94hh-jmq8-2fgp](https://github.com/apollographql/router/security/advisories/GHSA-94hh-jmq8-2fgp), and the `apollo-compiler` GitHub Advisory [GHSA-7mpv-9xg6-5r79](https://github.com/apollographql/apollo-rs/security/advisories/GHSA-7mpv-9xg6-5r79) for more information.
+
+By [@sachindshinde](https://github.com/sachindshinde) and [@goto-bus-stop](https://github.com/goto-bus-stop).
\ No newline at end of file