fix(schemas): Dont panic with non object field kinds

lib/vector-core/src/schema/definition.rs

@@ -265,6 +265,8 @@ impl Definition {
     /// A non-root required field means the root type must be an object, so the type will be automatically
     /// restricted to an object.
     ///
+    /// Raises a warning if the path is not root, and the definition does not allow the type to be an object.
+    ///
     /// # Panics
     /// - If the path is not root, and the definition does not allow the type to be an object.
     /// - Provided path has one or more coalesced segments (e.g. `.(foo | bar)`).
@@ -327,6 +329,8 @@ impl Definition {
     /// A non-root required field means the root type must be an object, so the type will be automatically
     /// restricted to an object.
     ///
+    /// Raises a warning if the path is not root, and the definition does not allow the type to be an object.
+    ///
     /// # Panics
     /// - If the path is not root, and the definition does not allow the type to be an object
     /// - Provided path has one or more coalesced segments (e.g. `.(foo | bar)`).

src/api/schema/components/mod.rs

@@ -14,14 +14,13 @@ use tokio_stream::{wrappers::BroadcastStream, Stream, StreamExt};
 use vector_config::NamedComponent;
 use vector_core::internal_event::DEFAULT_OUTPUT;
 
-use crate::topology::schema::possible_definitions;
 use crate::{
     api::schema::{
         components::state::component_by_component_key,
         filter::{self, filter_items},
         relay, sort,
     },
-    config::{ComponentKey, Config},
+    config::{transform_output_ids, ComponentKey, Config},
     filter_check,
 };
 
@@ -254,7 +253,6 @@ impl ComponentsSubscription {
 
 /// Update the 'global' configuration that will be consumed by component queries
 pub fn update_config(config: &Config) {
-    let mut cache = HashMap::new();
     let mut new_components = HashMap::new();
 
     // Sources
@@ -291,15 +289,13 @@ pub fn update_config(config: &Config) {
                 component_key: component_key.clone(),
                 component_type: transform.inner.get_component_name().to_string(),
                 inputs: transform.inputs.clone(),
-                outputs: transform
-                    .inner
-                    .outputs(
-                        &possible_definitions(&transform.inputs, config, &mut cache),
-                        config.schema.log_namespace(),
-                    )
-                    .into_iter()
-                    .map(|output| output.port.unwrap_or_else(|| DEFAULT_OUTPUT.to_string()))
-                    .collect(),
+                outputs: transform_output_ids(
+                    transform.inner.as_ref(),
+                    "".into(),
+                    config.schema.log_namespace(),
+                )
+                .map(|output| output.port.unwrap_or_else(|| DEFAULT_OUTPUT.to_string()))
+                .collect(),
             })),
         );
     }

src/config/compiler.rs

@@ -1,7 +1,8 @@
 use indexmap::IndexSet;
 
 use super::{
-    builder::ConfigBuilder, graph::Graph, id::Inputs, schema, validation, Config, OutputId,
+    builder::ConfigBuilder, graph::Graph, id::Inputs, transform::transform_output_ids, validation,
+    Config, OutputId,
 };
 
 pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<String>> {
@@ -137,16 +138,7 @@ pub(crate) fn expand_globs(config: &mut ConfigBuilder) {
                 })
         })
         .chain(config.transforms.iter().flat_map(|(key, t)| {
-            t.inner
-                .outputs(
-                    &[(key.into(), schema::Definition::any())],
-                    config.schema.log_namespace(),
-                )
-                .into_iter()
-                .map(|output| OutputId {
-                    component: key.clone(),
-                    port: output.port,
-                })
+            transform_output_ids(t.inner.as_ref(), key.clone(), config.schema.log_namespace())
         }))
         .map(|output_id| output_id.to_string())
         .collect::<IndexSet<String>>();

src/config/graph.rs

@@ -77,6 +77,7 @@ impl Graph {
                 Node::Transform {
                     in_ty: transform.inner.input().data_type(),
                     outputs: transform.inner.outputs(
+                        enrichment::TableRegistry::default(),
                         &[(id.into(), schema::Definition::any())],
                         schema.log_namespace(),
                     ),

src/config/mod.rs

@@ -54,7 +54,9 @@ pub use provider::ProviderConfig;
 pub use secret::SecretBackend;
 pub use sink::{SinkConfig, SinkContext, SinkHealthcheckOptions, SinkOuter};
 pub use source::{BoxedSource, SourceConfig, SourceContext, SourceOuter};
-pub use transform::{BoxedTransform, TransformConfig, TransformContext, TransformOuter};
+pub use transform::{
+    transform_output_ids, BoxedTransform, TransformConfig, TransformContext, TransformOuter,
+};
 pub use unit_test::{build_unit_tests, build_unit_tests_main, UnitTestResult};
 pub use validation::warnings;
 pub use vector_core::config::{

src/config/transform.rs

@@ -193,6 +193,7 @@ pub trait TransformConfig: DynClone + NamedComponent + core::fmt::Debug + Send +
     /// of events flowing through the transform.
     fn outputs(
         &self,
+        enrichment_tables: enrichment::TableRegistry,
         input_definitions: &[(OutputId, schema::Definition)],
         global_log_namespace: LogNamespace,
     ) -> Vec<TransformOutput>;
@@ -236,3 +237,23 @@ pub trait TransformConfig: DynClone + NamedComponent + core::fmt::Debug + Send +
 }
 
 dyn_clone::clone_trait_object!(TransformConfig);
+
+/// Often we want to call outputs just to retreive the OutputId's without needing
+/// the schema definitions.
+pub fn transform_output_ids<T: TransformConfig + ?Sized>(
+    transform: &T,
+    key: ComponentKey,
+    global_log_namespace: LogNamespace,
+) -> impl Iterator<Item = OutputId> + '_ {
+    transform
+        .outputs(
+            enrichment::TableRegistry::default(),
+            &[(key.clone().into(), schema::Definition::any())],
+            global_log_namespace,
+        )
+        .into_iter()
+        .map(move |output| OutputId {
+            component: key.clone(),
+            port: output.port,
+        })
+}

src/config/unit_test/mod.rs

@@ -20,7 +20,7 @@ pub use self::unit_test_components::{
     UnitTestSinkCheck, UnitTestSinkConfig, UnitTestSinkResult, UnitTestSourceConfig,
     UnitTestStreamSinkConfig, UnitTestStreamSourceConfig,
 };
-use super::{compiler::expand_globs, graph::Graph, OutputId};
+use super::{compiler::expand_globs, graph::Graph, transform::transform_output_ids, OutputId};
 use crate::{
     conditions::Condition,
     config::{
@@ -186,14 +186,11 @@ impl UnitTestBuildMetadata {
             .transforms
             .iter()
             .flat_map(|(key, transform)| {
-                transform
-                    .inner
-                    .outputs(&[], builder.schema.log_namespace())
-                    .into_iter()
-                    .map(|output| OutputId {
-                        component: key.clone(),
-                        port: output.port,
-                    })
+                transform_output_ids(
+                    transform.inner.as_ref(),
+                    key.clone(),
+                    builder.schema.log_namespace(),
+                )
             })
             .collect::<HashSet<_>>();
 
@@ -457,18 +454,13 @@ async fn build_unit_test(
 fn get_loose_end_outputs_sink(config: &ConfigBuilder) -> Option<SinkOuter<String>> {
     let config = config.clone();
     let transform_ids = config.transforms.iter().flat_map(|(key, transform)| {
-        transform
-            .inner
-            .outputs(&[], config.schema.log_namespace())
-            .iter()
-            .map(|output| {
-                if let Some(port) = &output.port {
-                    OutputId::from((key, port.clone())).to_string()
-                } else {
-                    key.to_string()
-                }
-            })
-            .collect::<Vec<_>>()
+        transform_output_ids(
+            transform.inner.as_ref(),
+            key.clone(),
+            config.schema.log_namespace(),
+        )
+        .map(|output| output.to_string())
+        .collect::<Vec<_>>()
     });
 
     let mut loose_end_outputs = Vec::new();

src/config/validation.rs

@@ -1,11 +1,14 @@
-use crate::{config::schema, topology::schema::possible_definitions};
+use crate::config::schema;
 use futures_util::{stream, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use heim::{disk::Partition, units::information::byte};
 use indexmap::IndexMap;
 use std::{collections::HashMap, path::PathBuf};
 use vector_core::internal_event::DEFAULT_OUTPUT;
 
-use super::{builder::ConfigBuilder, ComponentKey, Config, OutputId, Resource};
+use super::{
+    builder::ConfigBuilder, transform::transform_output_ids, ComponentKey, Config, OutputId,
+    Resource,
+};
 
 /// Check that provide + topology config aren't present in the same builder, which is an error.
 pub fn check_provider(config: &ConfigBuilder) -> Result<(), Vec<String>> {
@@ -169,15 +172,12 @@ pub fn check_outputs(config: &ConfigBuilder) -> Result<(), Vec<String>> {
             errors.extend(errs.into_iter().map(|msg| format!("Transform {key} {msg}")));
         }
 
-        if transform
-            .inner
-            .outputs(
-                &[(OutputId::dummy(), definition)],
-                config.schema.log_namespace(),
-            )
-            .iter()
-            .map(|output| output.port.as_deref().unwrap_or(""))
-            .any(|name| name == DEFAULT_OUTPUT)
+        if transform_output_ids(
+            transform.inner.as_ref(),
+            "dummy".into(),
+            config.schema.log_namespace(),
+        )
+        .any(|output| matches!(output.port, Some(output) if output == DEFAULT_OUTPUT))
         {
             errors.push(format!(
                 "Transform {key} cannot have a named output with reserved name: `{DEFAULT_OUTPUT}`"
@@ -325,7 +325,6 @@ async fn process_partitions(partitions: Vec<Partition>) -> heim::Result<IndexMap
 
 pub fn warnings(config: &Config) -> Vec<String> {
     let mut warnings = vec![];
-    let mut cache = HashMap::new();
 
     let source_ids = config.sources.iter().flat_map(|(key, source)| {
         source
@@ -342,21 +341,13 @@ pub fn warnings(config: &Config) -> Vec<String> {
             .collect::<Vec<_>>()
     });
     let transform_ids = config.transforms.iter().flat_map(|(key, transform)| {
-        transform
-            .inner
-            .outputs(
-                &possible_definitions(&transform.inputs, config, &mut cache),
-                config.schema.log_namespace(),
-            )
-            .iter()
-            .map(|output| {
-                if let Some(port) = &output.port {
-                    ("transform", OutputId::from((key, port.clone())))
-                } else {
-                    ("transform", OutputId::from(key))
-                }
-            })
-            .collect::<Vec<_>>()
+        transform_output_ids(
+            transform.inner.as_ref(),
+            key.clone(),
+            config.schema.log_namespace(),
+        )
+        .map(|output| ("transform", output))
+        .collect::<Vec<_>>()
     });
 
     for (input_type, id) in transform_ids.chain(source_ids) {

src/test_util/mock/transforms/basic.rs

@@ -51,6 +51,7 @@ impl TransformConfig for BasicTransformConfig {
 
     fn outputs(
         &self,
+        _: enrichment::TableRegistry,
         definitions: &[(OutputId, schema::Definition)],
         _: LogNamespace,
     ) -> Vec<TransformOutput> {

src/test_util/mock/transforms/noop.rs

@@ -41,6 +41,7 @@ impl TransformConfig for NoopTransformConfig {
 
     fn outputs(
         &self,
+        _: enrichment::TableRegistry,
         definitions: &[(OutputId, Definition)],
         _: LogNamespace,
     ) -> Vec<TransformOutput> {

src/topology/builder.rs

@@ -118,7 +118,7 @@ impl<'a> Builder<'a> {
         let enrichment_tables = self.load_enrichment_tables().await;
         let source_tasks = self.build_sources().await;
         self.build_transforms(enrichment_tables).await;
-        self.build_sinks().await;
+        self.build_sinks(enrichment_tables).await;
 
         // We should have all the data for the enrichment tables loaded now, so switch them over to
         // readonly.
@@ -402,8 +402,12 @@ impl<'a> Builder<'a> {
         {
             debug!(component = %key, "Building new transform.");
 
-            let input_definitions =
-                schema::input_definitions(&transform.inputs, self.config, &mut definition_cache);
+            let input_definitions = schema::input_definitions(
+                &transform.inputs,
+                self.config,
+                enrichment_tables.clone(),
+                &mut definition_cache,
+            );
 
             let merged_definition: Definition = input_definitions
                 .iter()
@@ -424,7 +428,11 @@ impl<'a> Builder<'a> {
             // Create a map of the outputs to the list of possible definitions from those outputs.
             let schema_definitions = transform
                 .inner
-                .outputs(&input_definitions, self.config.schema.log_namespace())
+                .outputs(
+                    enrichment_tables.clone(),
+                    &input_definitions,
+                    self.config.schema.log_namespace(),
+                )
                 .into_iter()
                 .map(|output| (output.port, output.log_schema_definitions))
                 .collect::<HashMap<_, _>>();
@@ -440,6 +448,7 @@ impl<'a> Builder<'a> {
 
             let node = TransformNode::from_parts(
                 key.clone(),
+                enrichment_tables.clone(),
                 transform,
                 &input_definitions,
                 self.config.schema.log_namespace(),
@@ -475,7 +484,7 @@ impl<'a> Builder<'a> {
         }
     }
 
-    async fn build_sinks(&mut self) {
+    async fn build_sinks(&mut self, enrichment_tables: &enrichment::TableRegistry) {
         for (key, sink) in self
             .config
             .sinks()
@@ -493,7 +502,12 @@ impl<'a> Builder<'a> {
             // At this point, we've validated that all transforms are valid, including any
             // transform that mutates the schema provided by their sources. We can now validate the
             // schema expectations of each individual sink.
-            if let Err(mut err) = schema::validate_sink_expectations(key, sink, self.config) {
+            if let Err(mut err) = schema::validate_sink_expectations(
+                key,
+                sink,
+                self.config,
+                enrichment_tables.clone(),
+            ) {
                 self.errors.append(&mut err);
             };
 
@@ -670,6 +684,7 @@ struct TransformNode {
 impl TransformNode {
     pub fn from_parts(
         key: ComponentKey,
+        enrichment_tables: enrichment::TableRegistry,
         transform: &TransformOuter<OutputId>,
         schema_definition: &[(OutputId, Definition)],
         global_log_namespace: LogNamespace,
@@ -679,9 +694,11 @@ impl TransformNode {
             typetag: transform.inner.get_component_name(),
             inputs: transform.inputs.clone(),
             input_details: transform.inner.input(),
-            outputs: transform
-                .inner
-                .outputs(schema_definition, global_log_namespace),
+            outputs: transform.inner.outputs(
+                enrichment_tables,
+                schema_definition,
+                global_log_namespace,
+            ),
             enable_concurrency: transform.inner.enable_concurrency(),
         }
     }