feat(federation): respect the subgraph_graphql_validation option

apollo-federation/src/operation/mod.rs

@@ -3934,7 +3934,7 @@ impl From<&FragmentSpreadSelection> for executable::FragmentSpread {
     }
 }
 
-impl TryFrom<Operation> for Valid<executable::ExecutableDocument> {
+impl TryFrom<Operation> for executable::ExecutableDocument {
     type Error = FederationError;
 
     fn try_from(value: Operation) -> Result<Self, Self::Error> {
@@ -3955,7 +3955,17 @@ impl TryFrom<Operation> for Valid<executable::ExecutableDocument> {
         document.fragments = fragments;
         document.operations.insert(operation);
         coerce_executable_values(value.schema.schema(), &mut document);
-        Ok(document.validate(value.schema.schema())?)
+        Ok(document)
+    }
+}
+
+impl TryFrom<Operation> for Valid<executable::ExecutableDocument> {
+    type Error = FederationError;
+
+    fn try_from(value: Operation) -> Result<Self, Self::Error> {
+        let schema = value.schema.clone();
+        let document: executable::ExecutableDocument = value.try_into()?;
+        Ok(document.validate(schema.schema())?)
     }
 }
 

apollo-federation/src/query_plan/mod.rs

@@ -6,6 +6,7 @@ use apollo_compiler::ExecutableDocument;
 use apollo_compiler::Name;
 use serde::Serialize;
 
+use crate::error::FederationError;
 use crate::query_plan::query_planner::QueryPlanningStatistics;
 
 pub(crate) mod conditions;
@@ -258,3 +259,121 @@ impl QueryPlan {
         }
     }
 }
+
+mod plan_node_visitors {
+    use super::*;
+
+    pub(super) fn fetch(
+        fetch: &FetchNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        callback(fetch)
+    }
+
+    pub(super) fn flatten(
+        flatten: &FlattenNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        flatten.node.visit_fetch_nodes(callback)
+    }
+
+    pub(super) fn sequence(
+        sequence: &SequenceNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        sequence
+            .nodes
+            .iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))
+    }
+
+    pub(super) fn parallel(
+        parallel: &ParallelNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        parallel
+            .nodes
+            .iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))
+    }
+
+    pub(super) fn condition(
+        condition: &ConditionNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        let ConditionNode {
+            if_clause,
+            else_clause,
+            ..
+        } = condition;
+        if_clause
+            .iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))?;
+        else_clause
+            .iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))?;
+        Ok(())
+    }
+
+    pub(super) fn subscription(
+        subscription: &SubscriptionNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        let SubscriptionNode { primary, rest } = subscription;
+        callback(primary)?;
+        rest.iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))?;
+        Ok(())
+    }
+
+    pub(super) fn defer(
+        defer: &DeferNode,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        let DeferNode { primary, deferred } = defer;
+        primary
+            .node
+            .iter()
+            .try_for_each(|node| node.visit_fetch_nodes(callback))?;
+        deferred
+            .iter()
+            .flat_map(|block| &block.node)
+            .try_for_each(|node| node.visit_fetch_nodes(callback))?;
+        Ok(())
+    }
+}
+
+impl PlanNode {
+    pub(crate) fn visit_fetch_nodes(
+        &self,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        match self {
+            Self::Fetch(fetch) => plan_node_visitors::fetch(fetch, callback),
+            Self::Sequence(sequence) => plan_node_visitors::sequence(sequence, callback),
+            Self::Parallel(parallel) => plan_node_visitors::parallel(parallel, callback),
+            Self::Flatten(flatten) => plan_node_visitors::flatten(flatten, callback),
+            Self::Condition(condition) => plan_node_visitors::condition(condition, callback),
+            Self::Defer(defer) => plan_node_visitors::defer(defer, callback),
+        }
+    }
+}
+
+impl TopLevelPlanNode {
+    pub(crate) fn visit_fetch_nodes(
+        &self,
+        callback: &mut impl FnMut(&FetchNode) -> Result<(), FederationError>,
+    ) -> Result<(), FederationError> {
+        match self {
+            Self::Fetch(fetch) => plan_node_visitors::fetch(fetch, callback),
+            Self::Sequence(sequence) => plan_node_visitors::sequence(sequence, callback),
+            Self::Parallel(parallel) => plan_node_visitors::parallel(parallel, callback),
+            Self::Flatten(flatten) => plan_node_visitors::flatten(flatten, callback),
+            Self::Condition(condition) => plan_node_visitors::condition(condition, callback),
+            Self::Defer(defer) => plan_node_visitors::defer(defer, callback),
+            Self::Subscription(subscription) => {
+                plan_node_visitors::subscription(subscription, callback)
+            }
+        }
+    }
+}

apollo-federation/src/query_plan/query_planner.rs

@@ -525,6 +525,22 @@ impl QueryPlanner {
             None => None,
         };
 
+        if self.config.subgraph_graphql_validation {
+            root_node.iter().try_for_each(|root_node| {
+                root_node.visit_fetch_nodes(&mut |node| {
+                    self.validate_fetch_node(node).map_err(|err| {
+                        let op_name = node
+                            .operation_name
+                            .as_ref()
+                            .map_or_else(|| "(anonymous)".to_string(), |name| name.to_string());
+                        FederationError::internal(format!(
+                            "Invalid fetch node operation: {op_name}\n{node}\n{err}",
+                        ))
+                    })
+                })
+            })?;
+        }
+
         let plan = QueryPlan {
             node: root_node,
             statistics,
@@ -540,6 +556,22 @@ impl QueryPlanner {
         &self.api_schema
     }
 
+    fn validate_fetch_node(&self, node: &FetchNode) -> Result<(), FederationError> {
+        let subgraph_schema = self
+            .subgraph_schemas()
+            .get(&node.subgraph_name)
+            .ok_or_else(|| {
+                FederationError::internal(format!(
+                    "subgraph schema not found: {}",
+                    node.subgraph_name
+                ))
+            })?;
+        // TODO: avoid cloning the operation document here
+        let document = node.operation_document.clone().into_inner();
+        document.validate(subgraph_schema.schema())?;
+        Ok(())
+    }
+
     fn check_unsupported_features(supergraph: &Supergraph) -> Result<(), FederationError> {
         // We have a *progressive* override when `join__field` has a
         // non-null value for `overrideLabel` field.