From 15a63b404262ef297929b3693d2387bfdd445f90 Mon Sep 17 00:00:00 2001
From: Nathan Fox <fuchsnj@gmail.com>
Date: Thu, 24 Aug 2023 13:06:54 -0400
Subject: [PATCH] fix(remap transform): log namespace should be used when
 splitting events from arrays (#18372)

fix remap array return values when using the Vector namespace
---
 lib/vector-core/src/event/vrl_target.rs | 12 ++++--
 src/transforms/remap.rs                 | 51 +++++++++++++++++++++++++
 2 files changed, 60 insertions(+), 3 deletions(-)

diff --git a/lib/vector-core/src/event/vrl_target.rs b/lib/vector-core/src/event/vrl_target.rs
index a3d039778b60c..8dd647e741c6a 100644
--- a/lib/vector-core/src/event/vrl_target.rs
+++ b/lib/vector-core/src/event/vrl_target.rs
@@ -50,6 +50,7 @@ pub struct TargetIter<T> {
     iter: std::vec::IntoIter<Value>,
     metadata: EventMetadata,
     _marker: PhantomData<T>,
+    log_namespace: LogNamespace,
 }
 
 fn create_log_event(value: Value, metadata: EventMetadata) -> LogEvent {
@@ -63,9 +64,12 @@ impl Iterator for TargetIter<LogEvent> {
 
     fn next(&mut self) -> Option<Self::Item> {
         self.iter.next().map(|v| {
-            match v {
-                value @ Value::Object(_) => LogEvent::from_parts(value, self.metadata.clone()),
-                value => create_log_event(value, self.metadata.clone()),
+            match self.log_namespace {
+                LogNamespace::Legacy => match v {
+                    value @ Value::Object(_) => LogEvent::from_parts(value, self.metadata.clone()),
+                    value => create_log_event(value, self.metadata.clone()),
+                },
+                LogNamespace::Vector => LogEvent::from_parts(v, self.metadata.clone()),
             }
             .into()
         })
@@ -142,6 +146,7 @@ impl VrlTarget {
                     iter: values.into_iter(),
                     metadata,
                     _marker: PhantomData,
+                    log_namespace,
                 }),
 
                 v => match log_namespace {
@@ -161,6 +166,7 @@ impl VrlTarget {
                     iter: values.into_iter(),
                     metadata,
                     _marker: PhantomData,
+                    log_namespace,
                 }),
 
                 v => TargetEvents::One(create_log_event(v, metadata).into()),
diff --git a/src/transforms/remap.rs b/src/transforms/remap.rs
index 94aa823d96463..18b5abe0e5516 100644
--- a/src/transforms/remap.rs
+++ b/src/transforms/remap.rs
@@ -1890,4 +1890,55 @@ mod tests {
             outputs1[0].schema_definitions(true),
         );
     }
+
+    #[test]
+    fn check_remap_array_vector_namespace() {
+        let event = {
+            let mut event = LogEvent::from("input");
+            // mark the event as a "Vector" namespaced log
+            event
+                .metadata_mut()
+                .value_mut()
+                .insert("vector", BTreeMap::new());
+            Event::from(event)
+        };
+
+        let conf = RemapConfig {
+            source: Some(
+                r#". = [null]
+"#
+                .to_string(),
+            ),
+            file: None,
+            drop_on_error: true,
+            drop_on_abort: false,
+            ..Default::default()
+        };
+        let mut tform = remap(conf.clone()).unwrap();
+        let result = transform_one(&mut tform, event).unwrap();
+
+        // Legacy namespace nests this under "message", Vector should set it as the root
+        assert_eq!(result.as_log().get("."), Some(&Value::Null));
+
+        let enrichment_tables = enrichment::TableRegistry::default();
+        let outputs1 = conf.outputs(
+            enrichment_tables,
+            &[(
+                "in".into(),
+                schema::Definition::new_with_default_metadata(
+                    Kind::any_object(),
+                    [LogNamespace::Vector],
+                ),
+            )],
+            LogNamespace::Vector,
+        );
+
+        let wanted =
+            schema::Definition::new_with_default_metadata(Kind::null(), [LogNamespace::Vector]);
+
+        assert_eq!(
+            HashMap::from([(OutputId::from("in"), wanted)]),
+            outputs1[0].schema_definitions(true),
+        );
+    }
 }