diff --git a/apollo-router/src/plugin/mod.rs b/apollo-router/src/plugin/mod.rs
index 0807a1856c..e38b1ca176 100644
--- a/apollo-router/src/plugin/mod.rs
+++ b/apollo-router/src/plugin/mod.rs
@@ -19,6 +19,7 @@ pub mod test;
 
 use std::any::TypeId;
 use std::fmt;
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
@@ -50,6 +51,7 @@ use crate::ListenAddr;
 type InstanceFactory = fn(
     &serde_json::Value,
     Arc<String>,
+    Arc<apollo_compiler::validation::Valid<apollo_compiler::Schema>>,
     Notify<String, graphql::Response>,
 ) -> BoxFuture<Result<Box<dyn DynPlugin>, BoxError>>;
 
@@ -66,6 +68,9 @@ pub struct PluginInit<T> {
     pub config: T,
     /// Router Supergraph Schema (schema definition language)
     pub supergraph_sdl: Arc<String>,
+    /// The supergraph schema (parsed)
+    pub(crate) supergraph_schema:
+        Arc<apollo_compiler::validation::Valid<apollo_compiler::schema::Schema>>,
 
     pub(crate) notify: Notify<String, graphql::Response>,
 }
@@ -79,6 +84,13 @@ where
     pub fn new(config: T, supergraph_sdl: Arc<String>) -> Self {
         Self::builder()
             .config(config)
+            .supergraph_schema(Arc::new(
+                apollo_compiler::schema::Schema::parse_and_validate(
+                    supergraph_sdl.to_string(),
+                    PathBuf::from("synthetic"),
+                )
+                .expect("failed to parse supergraph schema"),
+            ))
             .supergraph_sdl(supergraph_sdl)
             .notify(Notify::builder().build())
             .build()
@@ -95,6 +107,16 @@ where
     ) -> Result<Self, BoxError> {
         Self::try_builder()
             .config(config)
+            .supergraph_schema(Arc::new(
+                apollo_compiler::Schema::parse_and_validate(
+                    supergraph_sdl.to_string(),
+                    PathBuf::from("synthetic"),
+                )
+                .map_err(|e| {
+                    // This method is deprecated so we're not going to do anything fancy with the error
+                    BoxError::from(e.errors.to_string())
+                })?,
+            ))
             .supergraph_sdl(supergraph_sdl)
             .notify(Notify::builder().build())
             .build()
@@ -102,12 +124,31 @@ where
 
     #[cfg(test)]
     pub(crate) fn fake_new(config: T, supergraph_sdl: Arc<String>) -> Self {
+        let supergraph_schema = Arc::new(if !supergraph_sdl.is_empty() {
+            apollo_compiler::Schema::parse_and_validate(
+                supergraph_sdl.to_string(),
+                PathBuf::from("synthetic"),
+            )
+            .expect("failed to parse supergraph schema")
+        } else {
+            apollo_compiler::validation::Valid::assume_valid(apollo_compiler::Schema::new())
+        });
         PluginInit {
             config,
+            supergraph_schema,
             supergraph_sdl,
             notify: Notify::for_tests(),
         }
     }
+
+    /// Returns the parsed Schema. This is unstable and may be changed or removed in future router releases.
+    /// In addition, Schema is not stable, and may be changed or removed in future apollo-rs releases.
+    #[doc(hidden)]
+    pub fn unsupported_supergraph_schema(
+        &self,
+    ) -> Arc<apollo_compiler::validation::Valid<apollo_compiler::Schema>> {
+        self.supergraph_schema.clone()
+    }
 }
 
 #[buildstructor::buildstructor]
@@ -123,11 +164,13 @@ where
     pub(crate) fn new_builder(
         config: T,
         supergraph_sdl: Arc<String>,
+        supergraph_schema: Arc<apollo_compiler::validation::Valid<apollo_compiler::schema::Schema>>,
         notify: Notify<String, graphql::Response>,
     ) -> Self {
         PluginInit {
             config,
             supergraph_sdl,
+            supergraph_schema,
             notify,
         }
     }
@@ -140,12 +183,14 @@ where
     pub(crate) fn try_new_builder(
         config: serde_json::Value,
         supergraph_sdl: Arc<String>,
+        supergraph_schema: Arc<apollo_compiler::validation::Valid<apollo_compiler::schema::Schema>>,
         notify: Notify<String, graphql::Response>,
     ) -> Result<Self, BoxError> {
         let config: T = serde_json::from_value(config)?;
         Ok(PluginInit {
             config,
             supergraph_sdl,
+            supergraph_schema,
             notify,
         })
     }
@@ -155,11 +200,19 @@ where
     fn fake_new_builder(
         config: T,
         supergraph_sdl: Option<Arc<String>>,
+        supergraph_schema: Option<
+            Arc<apollo_compiler::validation::Valid<apollo_compiler::schema::Schema>>,
+        >,
         notify: Option<Notify<String, graphql::Response>>,
     ) -> Self {
         PluginInit {
             config,
             supergraph_sdl: supergraph_sdl.unwrap_or_default(),
+            supergraph_schema: supergraph_schema.unwrap_or_else(|| {
+                Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                    apollo_compiler::Schema::new(),
+                ))
+            }),
             notify: notify.unwrap_or_else(Notify::for_tests),
         }
     }
@@ -198,11 +251,12 @@ impl PluginFactory {
         tracing::debug!(%plugin_factory_name, "creating plugin factory");
         PluginFactory {
             name: plugin_factory_name,
-            instance_factory: |configuration, schema, notify| {
+            instance_factory: |configuration, schema, supergraph_schema, notify| {
                 Box::pin(async move {
                     let init = PluginInit::try_builder()
                         .config(configuration.clone())
                         .supergraph_sdl(schema)
+                        .supergraph_schema(supergraph_schema)
                         .notify(notify)
                         .build()?;
                     let plugin = P::new(init).await?;
@@ -225,11 +279,12 @@ impl PluginFactory {
         tracing::debug!(%plugin_factory_name, "creating plugin factory");
         PluginFactory {
             name: plugin_factory_name,
-            instance_factory: |configuration, schema, notify| {
+            instance_factory: |configuration, schema, supergraph_schema, notify| {
                 Box::pin(async move {
                     let init = PluginInit::try_builder()
                         .config(configuration.clone())
                         .supergraph_sdl(schema)
+                        .supergraph_schema(supergraph_schema)
                         .notify(notify)
                         .build()?;
                     let plugin = P::new(init).await?;
@@ -245,9 +300,10 @@ impl PluginFactory {
         &self,
         configuration: &serde_json::Value,
         supergraph_sdl: Arc<String>,
+        supergraph_schema: Arc<apollo_compiler::validation::Valid<apollo_compiler::Schema>>,
         notify: Notify<String, graphql::Response>,
     ) -> Result<Box<dyn DynPlugin>, BoxError> {
-        (self.instance_factory)(configuration, supergraph_sdl, notify).await
+        (self.instance_factory)(configuration, supergraph_sdl, supergraph_schema, notify).await
     }
 
     #[cfg(test)]
@@ -255,7 +311,15 @@ impl PluginFactory {
         &self,
         configuration: &serde_json::Value,
     ) -> Result<Box<dyn DynPlugin>, BoxError> {
-        (self.instance_factory)(configuration, Default::default(), Default::default()).await
+        (self.instance_factory)(
+            configuration,
+            Default::default(),
+            Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                apollo_compiler::Schema::new(),
+            )),
+            Default::default(),
+        )
+        .await
     }
 
     pub(crate) fn create_schema(&self, gen: &mut SchemaGenerator) -> schemars::schema::Schema {
diff --git a/apollo-router/src/plugins/override_url.rs b/apollo-router/src/plugins/override_url.rs
index 9d583024b2..4922a39f5f 100644
--- a/apollo-router/src/plugins/override_url.rs
+++ b/apollo-router/src/plugins/override_url.rs
@@ -113,7 +113,7 @@ mod tests {
         let dyn_plugin: Box<dyn DynPlugin> = crate::plugin::plugins()
             .find(|factory| factory.name == "apollo.override_subgraph_url")
             .expect("Plugin not found")
-            .create_instance(
+            .create_instance_without_schema(
                 &Value::from_str(
                     r#"{
                 "test_one": "http://localhost:8001",
@@ -121,8 +121,6 @@ mod tests {
             }"#,
                 )
                 .unwrap(),
-                Default::default(),
-                Default::default(),
             )
             .await
             .unwrap();
diff --git a/apollo-router/src/plugins/subscription.rs b/apollo-router/src/plugins/subscription.rs
index 1662e16cc9..202aca6e8c 100644
--- a/apollo-router/src/plugins/subscription.rs
+++ b/apollo-router/src/plugins/subscription.rs
@@ -681,6 +681,7 @@ fn ensure_id_consistency(
 #[cfg(test)]
 mod tests {
     use std::str::FromStr;
+    use std::sync::Arc;
 
     use futures::StreamExt;
     use serde_json::Value;
@@ -718,6 +719,9 @@ mod tests {
                 )
                 .unwrap(),
                 Default::default(),
+                Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                    apollo_compiler::Schema::new(),
+                )),
                 notify.clone(),
             )
             .await
@@ -856,6 +860,9 @@ mod tests {
                 )
                 .unwrap(),
                 Default::default(),
+                Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                    apollo_compiler::Schema::new(),
+                )),
                 notify.clone(),
             )
             .await
@@ -941,6 +948,9 @@ mod tests {
                 )
                 .unwrap(),
                 Default::default(),
+                Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                    apollo_compiler::Schema::new(),
+                )),
                 notify.clone(),
             )
             .await
@@ -1076,15 +1086,13 @@ mod tests {
         let dyn_plugin: Box<dyn DynPlugin> = crate::plugin::plugins()
             .find(|factory| factory.name == APOLLO_SUBSCRIPTION_PLUGIN)
             .expect("Plugin not found")
-            .create_instance(
+            .create_instance_without_schema(
                 &Value::from_str(
                     r#"{
                     "enabled": false
                 }"#,
                 )
                 .unwrap(),
-                Default::default(),
-                Default::default(),
             )
             .await
             .unwrap();
diff --git a/apollo-router/src/plugins/telemetry/logging/mod.rs b/apollo-router/src/plugins/telemetry/logging/mod.rs
index e417dad3c2..1064f28d35 100644
--- a/apollo-router/src/plugins/telemetry/logging/mod.rs
+++ b/apollo-router/src/plugins/telemetry/logging/mod.rs
@@ -2,7 +2,6 @@
 #[cfg(test)]
 mod test {
     use std::any::TypeId;
-    use std::sync::Arc;
 
     use tower::BoxError;
     use tower::ServiceBuilder;
@@ -11,8 +10,8 @@ mod test {
 
     use crate::assert_snapshot_subscriber;
     use crate::graphql;
+    use crate::plugin::DynPlugin;
     use crate::plugin::Plugin;
-    use crate::plugin::PluginInit;
     use crate::plugins::telemetry::Telemetry;
     use crate::services::router;
     use crate::services::subgraph;
@@ -151,39 +150,37 @@ mod test {
     // For now let's try and increase the coverage of the telemetry plugin using this and see how it goes.
 
     struct PluginTestHarness<T: Plugin> {
-        plugin: T,
+        plugin: Box<dyn DynPlugin>,
+        phantom: std::marker::PhantomData<T>,
     }
     #[buildstructor::buildstructor]
     impl<T: Plugin> PluginTestHarness<T> {
         #[builder]
-        async fn new(yaml: Option<&'static str>, supergraph: Option<&'static str>) -> Self {
+        async fn new(yaml: Option<&'static str>) -> Self {
             let factory = crate::plugin::plugins()
                 .find(|factory| factory.type_id == TypeId::of::<T>())
                 .expect("plugin not registered");
             let name = &factory.name.replace("apollo.", "");
             let config = yaml
-                .map(|yaml| serde_yaml::from_str::<serde_yaml::Value>(yaml).unwrap())
+                .map(|yaml| serde_yaml::from_str::<serde_json::Value>(yaml).unwrap())
                 .map(|mut config| {
                     config
-                        .as_mapping_mut()
+                        .as_object_mut()
                         .expect("invalid yaml")
-                        .remove(&serde_yaml::Value::String(name.to_string()))
+                        .remove(name)
                         .expect("no config for plugin")
                 })
-                .unwrap_or_else(|| serde_yaml::Value::Mapping(Default::default()));
-
-            let supergraph_sdl = supergraph
-                .map(|s| Arc::new(s.to_string()))
-                .unwrap_or_default();
-            let plugin = T::new(PluginInit {
-                config: serde_yaml::from_value(config).expect("config was invalid"),
-                supergraph_sdl,
-                notify: Default::default(),
-            })
-            .await
-            .expect("failed to initialize plugin");
-
-            Self { plugin }
+                .unwrap_or_else(|| serde_json::Value::Object(Default::default()));
+
+            let plugin = factory
+                .create_instance_without_schema(&config)
+                .await
+                .expect("failed to create plugin");
+
+            Self {
+                plugin,
+                phantom: Default::default(),
+            }
         }
 
         #[allow(dead_code)]
diff --git a/apollo-router/src/plugins/telemetry/mod.rs b/apollo-router/src/plugins/telemetry/mod.rs
index a030010ba9..05c409a013 100644
--- a/apollo-router/src/plugins/telemetry/mod.rs
+++ b/apollo-router/src/plugins/telemetry/mod.rs
@@ -1860,7 +1860,7 @@ mod tests {
         let mut plugin = crate::plugin::plugins()
             .find(|factory| factory.name == "apollo.telemetry")
             .expect("Plugin not found")
-            .create_instance(telemetry_config, Default::default(), Default::default())
+            .create_instance_without_schema(telemetry_config)
             .await
             .unwrap();
 
@@ -1935,10 +1935,8 @@ mod tests {
         crate::plugin::plugins()
             .find(|factory| factory.name == "apollo.telemetry")
             .expect("Plugin not found")
-            .create_instance(
+            .create_instance_without_schema(
                 &serde_json::json!({"apollo": {"schema_id":"abc"}, "exporters": {"tracing": {}}}),
-                Default::default(),
-                Default::default(),
             )
             .await
             .unwrap();
diff --git a/apollo-router/src/router_factory.rs b/apollo-router/src/router_factory.rs
index 1374fe7629..20f68c098a 100644
--- a/apollo-router/src/router_factory.rs
+++ b/apollo-router/src/router_factory.rs
@@ -157,7 +157,7 @@ impl RouterSuperServiceFactory for YamlRouterFactory {
         previous_router: Option<&'a Self::RouterFactory>,
         extra_plugins: Option<Vec<(String, Box<dyn DynPlugin>)>>,
     ) -> Result<Self::RouterFactory, BoxError> {
-        // we have to create afirst telemetry plugin before creating everything else, to generate a trace
+        // we have to create a telemetry plugin before creating everything else, to generate a trace
         // of router and plugin creation
         let plugin_registry = &*crate::plugin::PLUGINS;
         let mut initial_telemetry_plugin = None;
@@ -178,6 +178,9 @@ impl RouterSuperServiceFactory for YamlRouterFactory {
                         .create_instance(
                             plugin_config,
                             Arc::new(schema.clone()),
+                            Arc::new(apollo_compiler::validation::Valid::assume_valid(
+                                apollo_compiler::Schema::new(),
+                            )),
                             configuration.notify.clone(),
                         )
                         .await
@@ -485,6 +488,7 @@ pub(crate) async fn create_plugins(
     initial_telemetry_plugin: Option<Box<dyn DynPlugin>>,
     extra_plugins: Option<Vec<(String, Box<dyn DynPlugin>)>>,
 ) -> Result<Plugins, BoxError> {
+    let supergraph_schema = Arc::new(schema.definitions.clone());
     let mut apollo_plugins_config = configuration.apollo_plugins.clone().plugins;
     let user_plugins_config = configuration.plugins.clone().plugins.unwrap_or_default();
     let extra = extra_plugins.unwrap_or_default();
@@ -507,13 +511,14 @@ pub(crate) async fn create_plugins(
     let mut errors = Vec::new();
     let mut plugin_instances = Plugins::new();
 
-    // Use fonction-like macros to avoid borrow conflicts of captures
+    // Use function-like macros to avoid borrow conflicts of captures
     macro_rules! add_plugin {
         ($name: expr, $factory: expr, $plugin_config: expr) => {{
             match $factory
                 .create_instance(
                     &$plugin_config,
                     schema.as_string().clone(),
+                    supergraph_schema.clone(),
                     configuration.notify.clone(),
                 )
                 .await