Experimental: Introduce a pool of query planners

.changesets/exp_carton_ginger_magnet_beacon.md

@@ -0,0 +1,17 @@
+### Experimental: Introduce a pool of query planners ([PR #4897](https://github.com/apollographql/router/pull/4897))
+
+The router supports a new experimental feature: a pool of query planners to parallelize query planning.
+
+You can configure query planner pools with the `supergraph.query_planner.experimental_parallelism` option:
+
+```yaml
+supergraph:
+  query_planner:
+    experimental_parallelism: auto # number of available cpus
+```
+
+Its value is the number of query planners that run in parallel, and its default value is `1`. You can set it to the special value `auto` to automatically set it equal to the number of available CPUs.
+
+You can discuss and comment about query planner pools in this [GitHub discussion](https://github.com/apollographql/router/discussions/4917).
+
+By [@xuorig](https://github.com/xuorig) and [@o0Ignition0o](https://github.com/o0Ignition0o) in https://github.com/apollographql/router/pull/4897

Cargo.lock

@@ -231,7 +231,7 @@ dependencies = [
  "petgraph",
  "salsa",
  "serde_json",
- "strum 0.26.1",
+ "strum 0.26.2",
  "strum_macros 0.26.1",
  "thiserror",
  "url",
@@ -258,6 +258,7 @@ dependencies = [
  "apollo-federation",
  "arc-swap",
  "askama",
+ "async-channel 1.9.0",
  "async-compression",
  "async-trait",
  "aws-config",
@@ -6536,9 +6537,9 @@ dependencies = [
 
 [[package]]
 name = "strum"
-version = "0.26.1"
+version = "0.26.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "723b93e8addf9aa965ebe2d11da6d7540fa2283fcea14b3371ff055f7ba13f5f"
+checksum = "5d8cec3501a5194c432b2b7976db6b7d10ec95c253208b45f83f7136aa985e29"
 
 [[package]]
 name = "strum_macros"

apollo-router/Cargo.toml

@@ -67,6 +67,7 @@ anyhow = "1.0.80"
 apollo-compiler = "=1.0.0-beta.14"
 apollo-federation = "=0.0.9"
 arc-swap = "1.6.0"
+async-channel = "1.9.0"
 async-compression = { version = "0.4.6", features = [
     "tokio",
     "brotli",
@@ -350,3 +351,6 @@ harness = false
 [[bench]]
 name = "deeply_nested"
 harness = false
+
+[[example]]
+name = "planner"

apollo-router/examples/planner.rs

@@ -0,0 +1,66 @@
+use std::ops::ControlFlow;
+
+use anyhow::Result;
+use apollo_router::layers::ServiceBuilderExt;
+use apollo_router::plugin::Plugin;
+use apollo_router::plugin::PluginInit;
+use apollo_router::register_plugin;
+use apollo_router::services::execution;
+use apollo_router::services::supergraph;
+use tower::BoxError;
+use tower::ServiceBuilder;
+use tower::ServiceExt;
+
+#[derive(Debug)]
+struct DoNotExecute {
+    #[allow(dead_code)]
+    configuration: bool,
+}
+
+#[async_trait::async_trait]
+impl Plugin for DoNotExecute {
+    type Config = bool;
+
+    async fn new(init: PluginInit<Self::Config>) -> Result<Self, BoxError> {
+        Ok(Self {
+            configuration: init.config,
+        })
+    }
+
+    fn supergraph_service(&self, service: supergraph::BoxService) -> supergraph::BoxService {
+        ServiceBuilder::new()
+            .map_request(|mut req: supergraph::Request| {
+                let body = req.supergraph_request.body_mut();
+                body.query = body.query.as_ref().map(|query| {
+                    let query_name = format!("query Query{} ", rand::random::<usize>());
+                    query.replacen("query ", query_name.as_str(), 1)
+                });
+                req
+            })
+            .service(service)
+            .boxed()
+    }
+
+    fn execution_service(&self, service: execution::BoxService) -> execution::BoxService {
+        ServiceBuilder::new()
+            .checkpoint(|req: execution::Request| {
+                Ok(ControlFlow::Break(
+                    execution::Response::fake_builder()
+                        .context(req.context)
+                        .build()
+                        .unwrap(),
+                ))
+            })
+            .service(service)
+            .boxed()
+    }
+}
+
+register_plugin!("apollo-test", "do_not_execute", DoNotExecute);
+
+// Run this benchmark with cargo run --release --example planner -- --hot-reload -s <path/to/your/schema.graphql> -c ./apollo-router/examples/router.yaml
+// You can then send operations to it with `ab` or `hey` or any tool you like:
+// hey -n 1000 -c 10 -m POST -H 'Content-Type: application/json' -D 'path/to/an/anonymous/operation' http://localhost:4100
+fn main() -> Result<()> {
+    apollo_router::main()
+}

apollo-router/examples/router.yaml

@@ -0,0 +1,13 @@
+supergraph:
+  listen: 0.0.0.0:4100
+  introspection: true
+  query_planner:
+    experimental_parallelism: auto # or any number
+plugins:
+  experimental.expose_query_plan: true
+  apollo-test.do_not_execute: true
+experimental_graphql_validation_mode: both
+sandbox:
+  enabled: true
+homepage:
+  enabled: false

apollo-router/src/axum_factory/tests.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::io;
 use std::net::SocketAddr;
+use std::num::NonZeroUsize;
 use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::atomic::AtomicU32;
@@ -64,7 +65,7 @@ use crate::http_server_factory::HttpServerFactory;
 use crate::http_server_factory::HttpServerHandle;
 use crate::json_ext::Path;
 use crate::plugin::test::MockSubgraph;
-use crate::query_planner::BridgeQueryPlanner;
+use crate::query_planner::BridgeQueryPlannerPool;
 use crate::router_factory::create_plugins;
 use crate::router_factory::Endpoint;
 use crate::router_factory::RouterFactory;
@@ -2303,9 +2304,13 @@ async fn test_supergraph_timeout() {
     let conf: Arc<Configuration> = Arc::new(serde_json::from_value(config).unwrap());
 
     let schema = include_str!("..//testdata/minimal_supergraph.graphql");
-    let planner = BridgeQueryPlanner::new(schema.to_string(), conf.clone())
-        .await
-        .unwrap();
+    let planner = BridgeQueryPlannerPool::new(
+        schema.to_string(),
+        conf.clone(),
+        NonZeroUsize::new(1).unwrap(),
+    )
+    .await
+    .unwrap();
     let schema = planner.schema();
 
     // we do the entire supergraph rebuilding instead of using `from_supergraph_mock_callback_and_configuration`

apollo-router/src/configuration/metrics.rs

@@ -8,6 +8,7 @@ use opentelemetry_api::KeyValue;
 use paste::paste;
 use serde_json::Value;
 
+use super::AvailableParallelism;
 use crate::metrics::meter_provider;
 use crate::uplink::license_enforcement::LicenseState;
 use crate::Configuration;
@@ -46,6 +47,7 @@ impl Metrics {
         );
         data.populate_license_instrument(license_state);
         data.populate_user_plugins_instrument(configuration);
+        data.populate_query_planner_experimental_parallelism(configuration);
         data.into()
     }
 }
@@ -436,6 +438,45 @@ impl InstrumentData {
             ),
         );
     }
+
+    pub(crate) fn populate_query_planner_experimental_parallelism(
+        &mut self,
+        configuration: &Configuration,
+    ) {
+        let query_planner_parallelism_config = configuration
+            .supergraph
+            .query_planner
+            .experimental_parallelism;
+
+        if query_planner_parallelism_config != Default::default() {
+            let mut attributes = HashMap::new();
+            attributes.insert(
+                "mode".to_string(),
+                if let AvailableParallelism::Auto(_) = query_planner_parallelism_config {
+                    "auto"
+                } else {
+                    "static"
+                }
+                .into(),
+            );
+            self.data.insert(
+                "apollo.router.config.query_panner.parallelism".to_string(),
+                (
+                    configuration
+                        .supergraph
+                        .query_planner
+                        .experimental_query_planner_parallelism()
+                        .map(|n| {
+                            let as_usize: usize = n.into();
+                            let as_u64: u64 = as_usize.try_into().unwrap_or_default();
+                            as_u64
+                        })
+                        .unwrap_or_default(),
+                    attributes,
+                ),
+            );
+        }
+    }
 }
 
 impl From<InstrumentData> for Metrics {
@@ -469,6 +510,7 @@ mod test {
     use crate::configuration::metrics::InstrumentData;
     use crate::configuration::metrics::Metrics;
     use crate::uplink::license_enforcement::LicenseState;
+    use crate::Configuration;
 
     #[derive(RustEmbed)]
     #[folder = "src/configuration/testdata/metrics"]
@@ -486,6 +528,8 @@ mod test {
 
             let mut data = InstrumentData::default();
             data.populate_config_instruments(yaml);
+            let configuration: Configuration = input.parse().unwrap();
+            data.populate_query_planner_experimental_parallelism(&configuration);
             let _metrics: Metrics = data.into();
             assert_non_zero_metrics_snapshot!(file_name);
         }

apollo-router/src/configuration/mod.rs

@@ -629,6 +629,46 @@ pub(crate) struct Supergraph {
     /// Log a message if the client closes the connection before the response is sent.
     /// Default: false.
     pub(crate) experimental_log_on_broken_pipe: bool,
+
+    /// Configuration options pertaining to the query planner component.
+    pub(crate) query_planner: QueryPlanner,
+}
+
+/// Configuration options pertaining to the query planner component.
+#[derive(Debug, Clone, Default, Deserialize, Serialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct QueryPlanner {
+    /// Set the size of a pool of workers to enable query planning parallelism.
+    /// Default: 1.
+    pub(crate) experimental_parallelism: AvailableParallelism,
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
+#[serde(rename_all = "snake_case", untagged)]
+pub(crate) enum AvailableParallelism {
+    Auto(Auto),
+    Fixed(NonZeroUsize),
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum Auto {
+    Auto,
+}
+
+impl Default for AvailableParallelism {
+    fn default() -> Self {
+        Self::Fixed(NonZeroUsize::new(1).expect("cannot fail"))
+    }
+}
+
+impl QueryPlanner {
+    pub(crate) fn experimental_query_planner_parallelism(&self) -> io::Result<NonZeroUsize> {
+        match self.experimental_parallelism {
+            AvailableParallelism::Auto(Auto::Auto) => std::thread::available_parallelism(),
+            AvailableParallelism::Fixed(n) => Ok(n),
+        }
+    }
 }
 
 fn default_defer_support() -> bool {
@@ -648,6 +688,7 @@ impl Supergraph {
         generate_query_fragments: Option<bool>,
         early_cancel: Option<bool>,
         experimental_log_on_broken_pipe: Option<bool>,
+        query_planner: Option<QueryPlanner>,
     ) -> Self {
         Self {
             listen: listen.unwrap_or_else(default_graphql_listen),
@@ -667,6 +708,7 @@ impl Supergraph {
             generate_query_fragments: generate_query_fragments.unwrap_or_default(),
             early_cancel: early_cancel.unwrap_or_default(),
             experimental_log_on_broken_pipe: experimental_log_on_broken_pipe.unwrap_or_default(),
+            query_planner: query_planner.unwrap_or_default(),
         }
     }
 }
@@ -685,6 +727,7 @@ impl Supergraph {
         generate_query_fragments: Option<bool>,
         early_cancel: Option<bool>,
         experimental_log_on_broken_pipe: Option<bool>,
+        query_planner: Option<QueryPlanner>,
     ) -> Self {
         Self {
             listen: listen.unwrap_or_else(test_listen),
@@ -704,6 +747,7 @@ impl Supergraph {
             generate_query_fragments: generate_query_fragments.unwrap_or_default(),
             early_cancel: early_cancel.unwrap_or_default(),
             experimental_log_on_broken_pipe: experimental_log_on_broken_pipe.unwrap_or_default(),
+            query_planner: query_planner.unwrap_or_default(),
         }
     }
 }

apollo-router/src/configuration/snapshots/apollo_router__configuration__metrics__test__metrics@query_planner_parallelism_auto.router.yaml.snap

@@ -0,0 +1,10 @@
+---
+source: apollo-router/src/configuration/metrics.rs
+expression: "&metrics.non_zero()"
+---
+- name: apollo.router.config.query_panner.parallelism
+  data:
+    datapoints:
+      - value: 8
+        attributes:
+          mode: auto

apollo-router/src/configuration/snapshots/apollo_router__configuration__metrics__test__metrics@query_planner_parallelism_static.router.yaml.snap

@@ -0,0 +1,10 @@
+---
+source: apollo-router/src/configuration/metrics.rs
+expression: "&metrics.non_zero()"
+---
+- name: apollo.router.config.query_panner.parallelism
+  data:
+    datapoints:
+      - value: 10
+        attributes:
+          mode: static