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

lib/vector-core/src/config/mod.rs

@@ -201,7 +201,7 @@ pub struct TransformOutput {
     /// enabled, at least one definition  should be output. If the transform
     /// has multiple connected sources, it is possible to have multiple output
     /// definitions - one for each input.
-    pub log_schema_definitions: HashMap<OutputId, schema::Definition>,
+    log_schema_definitions: HashMap<OutputId, schema::Definition>,
 }
 
 impl TransformOutput {
@@ -222,6 +222,37 @@ impl TransformOutput {
         self.port = Some(name.into());
         self
     }
+
+    /// Return the schema [`schema::Definition`] from this output.
+    ///
+    /// Takes a `schema_enabled` flag to determine if the full definition including the fields
+    /// and associated types should be returned, or if a simple definition should be returned.
+    /// A simple definition is just the default for the namespace. For the Vector namespace the
+    /// meanings are included.
+    /// Schema enabled is set in the users configuration.
+    #[must_use]
+    pub fn schema_definitions(
+        &self,
+        schema_enabled: bool,
+    ) -> HashMap<OutputId, schema::Definition> {
+        if schema_enabled {
+            self.log_schema_definitions.clone()
+        } else {
+            self.log_schema_definitions
+                .iter()
+                .map(|(output, definition)| {
+                    let mut new_definition =
+                        schema::Definition::default_for_namespace(definition.log_namespaces());
+
+                    if definition.log_namespaces().contains(&LogNamespace::Vector) {
+                        new_definition.add_meanings(definition.meanings());
+                    }
+
+                    (output.clone(), new_definition)
+                })
+                .collect()
+        }
+    }
 }
 
 /// Simple utility function that can be used by transforms that make no changes to

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::{get_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: get_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::get_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,
-                })
+            get_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::{
+    get_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 retrieve the OutputId's without needing
+/// the schema definitions.
+pub fn get_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::get_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,
-                    })
+                get_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<_>>()
+        get_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::get_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 get_transform_output_ids(
+            transform.inner.as_ref(),
+            key.clone(),
+            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<_>>()
+        get_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/mod.rs

@@ -16,7 +16,7 @@ use self::{
         BackpressureSourceConfig, BasicSourceConfig, ErrorSourceConfig, PanicSourceConfig,
         TripwireSourceConfig,
     },
-    transforms::BasicTransformConfig,
+    transforms::{BasicTransformConfig, ErrorDefinitionTransformConfig},
 };
 
 pub mod sinks;
@@ -66,6 +66,10 @@ pub fn basic_transform(suffix: &str, increase: f64) -> BasicTransformConfig {
     BasicTransformConfig::new(suffix.to_owned(), increase)
 }
 
+pub const fn error_definition_transform() -> ErrorDefinitionTransformConfig {
+    ErrorDefinitionTransformConfig {}
+}
+
 pub const fn backpressure_sink(num_to_consume: usize) -> BackpressureSinkConfig {
     BackpressureSinkConfig { num_to_consume }
 }

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/error_definitions.rs

@@ -0,0 +1,62 @@
+use async_trait::async_trait;
+use snafu::Snafu;
+use value::Kind;
+use vector_config::configurable_component;
+use vector_core::{
+    config::{DataType, Input, LogNamespace, TransformOutput},
+    schema::Definition,
+    transform::Transform,
+};
+
+use crate::config::{OutputId, TransformConfig, TransformContext};
+
+#[derive(Debug, Snafu)]
+enum Error {
+    #[snafu(display("It all went horribly wrong"))]
+    ItAllWentHorriblyWrong,
+}
+
+/// Configuration for the `test_error_definition` transform.
+#[configurable_component(transform("test_error_definition", "Test (error definition)"))]
+#[derive(Clone, Debug, Default)]
+pub struct ErrorDefinitionTransformConfig {}
+
+impl_generate_config_from_default!(ErrorDefinitionTransformConfig);
+
+#[async_trait]
+#[typetag::serde(name = "test_error_definition")]
+impl TransformConfig for ErrorDefinitionTransformConfig {
+    fn input(&self) -> Input {
+        Input::all()
+    }
+
+    fn outputs(
+        &self,
+        _: enrichment::TableRegistry,
+        definitions: &[(OutputId, Definition)],
+        _: LogNamespace,
+    ) -> Vec<TransformOutput> {
+        vec![TransformOutput::new(
+            DataType::all(),
+            definitions
+                .iter()
+                .map(|(output, definition)| {
+                    (
+                        output.clone(),
+                        // Return a definition of Kind::never implying that we can never return a value.
+                        Definition::new_with_default_metadata(
+                            Kind::never(),
+                            definition.log_namespaces().clone(),
+                        ),
+                    )
+                })
+                .collect(),
+        )]
+    }
+
+    async fn build(&self, _: &TransformContext) -> crate::Result<Transform> {
+        // Even though the definitions returned were `Kind::never`, build needs to be
+        // called in order to return the Error.
+        Err(Error::ItAllWentHorriblyWrong.into())
+    }
+}