Revert "chore(topology): Transform outputs hash table of OutputId -> Definition (#17059)"

This reverts commit 1bdb24d.

Author: StephenWakely

benches/remap.rs

@@ -28,7 +28,7 @@ fn benchmark_remap(c: &mut Criterion) {
 
     let add_fields_runner = |tform: &mut Box<dyn SyncTransform>, event: Event| {
         let mut outputs = TransformOutputsBuf::new_with_capacity(
-            vec![TransformOutput::new(DataType::all(), HashMap::new())],
+            vec![TransformOutput::new(DataType::all(), vec![])],
             1,
         );
         tform.transform(event, &mut outputs);
@@ -80,7 +80,7 @@ fn benchmark_remap(c: &mut Criterion) {
 
     let json_parser_runner = |tform: &mut Box<dyn SyncTransform>, event: Event| {
         let mut outputs = TransformOutputsBuf::new_with_capacity(
-            vec![TransformOutput::new(DataType::all(), HashMap::new())],
+            vec![TransformOutput::new(DataType::all(), vec![])],
             1,
         );
         tform.transform(event, &mut outputs);
@@ -134,7 +134,7 @@ fn benchmark_remap(c: &mut Criterion) {
     let coerce_runner =
         |tform: &mut Box<dyn SyncTransform>, event: Event, timestamp: DateTime<Utc>| {
             let mut outputs = TransformOutputsBuf::new_with_capacity(
-                vec![TransformOutput::new(DataType::all(), HashMap::new())],
+                vec![TransformOutput::new(DataType::all(), vec![])],
                 1,
             );
             tform.transform(event, &mut outputs);

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

@@ -1,19 +1,17 @@
-use std::{collections::HashMap, fmt, num::NonZeroUsize};
+use std::{fmt, num::NonZeroUsize};
 
 use bitmask_enum::bitmask;
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
 
 mod global_options;
 mod log_schema;
-pub mod output_id;
 pub mod proxy;
 
 use crate::event::LogEvent;
 pub use global_options::GlobalOptions;
 pub use log_schema::{init_log_schema, log_schema, LogSchema};
 use lookup::{lookup_v2::ValuePath, path, PathPrefix};
-pub use output_id::OutputId;
 use serde::{Deserialize, Serialize};
 use value::Value;
 pub use vector_common::config::ComponentKey;
@@ -201,14 +199,14 @@ 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>,
+    pub log_schema_definitions: Vec<schema::Definition>,
 }
 
 impl TransformOutput {
     /// Create a `TransformOutput` of the given data type that contains multiple [`schema::Definition`]s.
     /// Designed for use in transforms.
     #[must_use]
-    pub fn new(ty: DataType, schema_definitions: HashMap<OutputId, schema::Definition>) -> Self {
+    pub fn new(ty: DataType, schema_definitions: Vec<schema::Definition>) -> Self {
         Self {
             port: None,
             ty,
@@ -224,18 +222,6 @@ impl TransformOutput {
     }
 }
 
-/// Simple utility function that can be used by transforms that make no changes to
-/// the schema definitions of events.
-/// Takes a list of definitions with [`OutputId`] returns them as a [`HashMap`].
-pub fn clone_input_definitions(
-    input_definitions: &[(OutputId, schema::Definition)],
-) -> HashMap<OutputId, schema::Definition> {
-    input_definitions
-        .iter()
-        .map(|(output, definition)| (output.clone(), definition.clone()))
-        .collect()
-}
-
 /// Source-specific end-to-end acknowledgements configuration.
 ///
 /// This type exists solely to provide a source-specific description of the `acknowledgements`

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

@@ -399,7 +399,7 @@ mod test {
                     in_ty,
                     outputs: vec![TransformOutput::new(
                         out_ty,
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
+                        vec![Definition::default_legacy_namespace()],
                     )],
                 },
             );
@@ -415,11 +415,8 @@ mod test {
             let id = id.into();
             match self.nodes.get_mut(&id) {
                 Some(Node::Transform { outputs, .. }) => outputs.push(
-                    TransformOutput::new(
-                        ty,
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
-                    )
-                    .with_port(name),
+                    TransformOutput::new(ty, vec![Definition::default_legacy_namespace()])
+                        .with_port(name),
                 ),
                 _ => panic!("invalid transform"),
             }
@@ -654,11 +651,11 @@ mod test {
                 outputs: vec![
                     TransformOutput::new(
                         DataType::all(),
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
+                        vec![Definition::default_legacy_namespace()],
                     ),
                     TransformOutput::new(
                         DataType::all(),
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
+                        vec![Definition::default_legacy_namespace()],
                     )
                     .with_port("bar"),
                 ],
@@ -679,11 +676,11 @@ mod test {
                 outputs: vec![
                     TransformOutput::new(
                         DataType::all(),
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
+                        vec![Definition::default_legacy_namespace()],
                     ),
                     TransformOutput::new(
                         DataType::all(),
-                        [("test".into(), Definition::default_legacy_namespace())].into(),
+                        vec![Definition::default_legacy_namespace()],
                     )
                     .with_port("errors"),
                 ],

src/config/graph.rs

@@ -1,8 +1,10 @@
-use std::ops::Deref;
+use std::{fmt, ops::Deref};
 
 use vector_config::configurable_component;
 pub use vector_core::config::ComponentKey;
 
+use super::schema;
+
 /// A list of upstream [source][sources] or [transform][transforms] IDs.
 ///
 /// Wildcards (`*`) are supported.
@@ -94,3 +96,88 @@ impl<T> From<Vec<T>> for Inputs<T> {
         Self(inputs)
     }
 }
+
+/// Component output identifier.
+#[configurable_component]
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct OutputId {
+    /// The component to which the output belongs.
+    pub component: ComponentKey,
+
+    /// The output port name, if not the default.
+    pub port: Option<String>,
+}
+
+impl OutputId {
+    /// Some situations, for example when validating a config file requires running the
+    /// transforms::output function to retrieve the outputs, but we don't have an
+    /// `OutputId` from a source. This gives us an `OutputId` that we can use.
+    ///
+    /// TODO: This is not a pleasant solution, but would require some significant refactoring
+    /// to the topology code to avoid.
+    pub fn dummy() -> Self {
+        Self {
+            component: "dummy".into(),
+            port: None,
+        }
+    }
+
+    /// Given a list of [`schema::Definition`]s, returns a [`Vec`] of tuples of
+    /// this `OutputId` with each `Definition`.
+    pub fn with_definitions(
+        &self,
+        definitions: impl IntoIterator<Item = schema::Definition>,
+    ) -> Vec<(OutputId, schema::Definition)> {
+        definitions
+            .into_iter()
+            .map(|definition| (self.clone(), definition))
+            .collect()
+    }
+}
+
+impl fmt::Display for OutputId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match &self.port {
+            None => self.component.fmt(f),
+            Some(port) => write!(f, "{}.{}", self.component, port),
+        }
+    }
+}
+
+impl From<ComponentKey> for OutputId {
+    fn from(key: ComponentKey) -> Self {
+        Self {
+            component: key,
+            port: None,
+        }
+    }
+}
+
+impl From<&ComponentKey> for OutputId {
+    fn from(key: &ComponentKey) -> Self {
+        Self::from(key.clone())
+    }
+}
+
+impl From<(&ComponentKey, String)> for OutputId {
+    fn from((key, name): (&ComponentKey, String)) -> Self {
+        Self {
+            component: key.clone(),
+            port: Some(name),
+        }
+    }
+}
+
+// This panicking implementation is convenient for testing, but should never be enabled for use
+// outside of tests.
+#[cfg(test)]
+impl From<&str> for OutputId {
+    fn from(s: &str) -> Self {
+        assert!(
+            !s.contains('.'),
+            "Cannot convert dotted paths to strings without more context"
+        );
+        let component = ComponentKey::from(s);
+        component.into()
+    }
+}

src/config/id.rs

@@ -45,7 +45,7 @@ pub use cmd::{cmd, Opts};
 pub use diff::ConfigDiff;
 pub use enrichment_table::{EnrichmentTableConfig, EnrichmentTableOuter};
 pub use format::{Format, FormatHint};
-pub use id::{ComponentKey, Inputs};
+pub use id::{ComponentKey, Inputs, OutputId};
 pub use loading::{
     load, load_builder_from_paths, load_from_paths, load_from_paths_with_provider_and_secrets,
     load_from_str, load_source_from_paths, merge_path_lists, process_paths, CONFIG_PATHS,
@@ -57,9 +57,7 @@ pub use source::{BoxedSource, SourceConfig, SourceContext, SourceOuter};
 pub use transform::{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::{
-    init_log_schema, log_schema, proxy::ProxyConfig, LogSchema, OutputId,
-};
+pub use vector_core::config::{init_log_schema, log_schema, proxy::ProxyConfig, LogSchema};
 
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq)]
 pub enum ConfigPath {

src/config/mod.rs

@@ -111,7 +111,7 @@ pub struct TransformContext {
     ///
     /// Given a transform can expose multiple [`TransformOutput`] channels, the ID is tied to the identifier of
     /// that `TransformOutput`.
-    pub schema_definitions: HashMap<Option<String>, HashMap<OutputId, schema::Definition>>,
+    pub schema_definitions: HashMap<Option<String>, Vec<schema::Definition>>,
 
     /// The schema definition created by merging all inputs of the transform.
     ///
@@ -129,7 +129,7 @@ impl Default for TransformContext {
             key: Default::default(),
             globals: Default::default(),
             enrichment_tables: Default::default(),
-            schema_definitions: HashMap::from([(None, HashMap::new())]),
+            schema_definitions: HashMap::from([(None, vec![schema::Definition::any()])]),
             merged_schema_definition: schema::Definition::any(),
             schema: SchemaOptions::default(),
         }
@@ -148,9 +148,7 @@ impl TransformContext {
     }
 
     #[cfg(any(test, feature = "test"))]
-    pub fn new_test(
-        schema_definitions: HashMap<Option<String>, HashMap<OutputId, schema::Definition>>,
-    ) -> Self {
+    pub fn new_test(schema_definitions: HashMap<Option<String>, Vec<schema::Definition>>) -> Self {
         Self {
             schema_definitions,
             ..Default::default()

src/config/transform.rs

@@ -58,7 +58,7 @@ impl TransformConfig for BasicTransformConfig {
             DataType::all(),
             definitions
                 .iter()
-                .map(|(output, definition)| (output.clone(), definition.clone()))
+                .map(|(_output, definition)| definition.clone())
                 .collect(),
         )]
     }

src/test_util/mock/transforms/basic.rs

@@ -48,7 +48,7 @@ impl TransformConfig for NoopTransformConfig {
             DataType::all(),
             definitions
                 .iter()
-                .map(|(output, definition)| (output.clone(), definition.clone()))
+                .map(|(_output, definition)| definition.clone())
                 .collect(),
         )]
     }