enhancement(vrl): Update to use VRL with Copy-on-Write optimization

Cargo.toml

@@ -171,7 +171,8 @@ vector-lib = { path = "lib/vector-lib", default-features = false, features = ["v
 vector-config = { path = "lib/vector-config" }
 vector-config-common = { path = "lib/vector-config-common" }
 vector-config-macros = { path = "lib/vector-config-macros" }
-vrl = { git = "https://github.com/vectordotdev/vrl", branch = "main", features = ["arbitrary", "cli", "test", "test_framework"] }
+# TODO: change when https://github.com/vectordotdev/vrl/pull/1349 is merged
+vrl = { git = "https://github.com/Zettroke/vrl", branch = "CoW-value-optimization", features = ["arbitrary", "cli", "test", "test_framework"] }
 
 [dependencies]
 pin-project.workspace = true

lib/codecs/src/decoding/format/avro.rs

@@ -187,7 +187,7 @@ pub fn try_from(value: AvroValue) -> vector_common::Result<VrlValue> {
             for item in array {
                 vector.push(try_from(item)?);
             }
-            Ok(VrlValue::Array(vector))
+            Ok(VrlValue::Array(vector.into()))
         }
         AvroValue::Boolean(boolean) => Ok(VrlValue::from(boolean)),
         AvroValue::Bytes(bytes) => Ok(VrlValue::from(bytes)),

lib/codecs/src/encoding/format/gelf.rs

@@ -236,15 +236,16 @@ fn to_gelf_event(log: LogEvent) -> vector_common::Result<LogEvent> {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
     use crate::encoding::SerializerConfig;
 
     use super::*;
     use chrono::NaiveDateTime;
     use vector_core::event::{Event, EventMetadata};
     use vrl::btreemap;
-    use vrl::value::{ObjectMap, Value};
+    use vrl::value::Value;
 
-    fn do_serialize(expect_success: bool, event_fields: ObjectMap) -> Option<serde_json::Value> {
+    fn do_serialize(expect_success: bool, event_fields: BTreeMap<KeyString, Value>) -> Option<serde_json::Value> {
         let config = GelfSerializerConfig::new();
         let mut serializer = config.build();
         let event: Event = LogEvent::from_map(event_fields, EventMetadata::default()).into();

lib/opentelemetry-proto/src/convert.rs

@@ -7,7 +7,7 @@ use vector_core::{
     config::{log_schema, LegacyKey, LogNamespace},
     event::{Event, LogEvent, TraceEvent},
 };
-use vrl::value::KeyString;
+use vrl::value::{KeyString, Array};
 use vrl::{
     event_path,
     value::{ObjectMap, Value},
@@ -87,7 +87,7 @@ impl From<PBValue> for Value {
                 arr.values
                     .into_iter()
                     .map(|av| av.value.map(Into::into).unwrap_or(Value::Null))
-                    .collect::<Vec<Value>>(),
+                    .collect::<Array>(),
             ),
             PBValue::KvlistValue(arr) => kv_list_into_value(arr.values),
         }
@@ -395,7 +395,7 @@ impl From<SpanEvent> for Value {
             "dropped_attributes_count".into(),
             Value::Integer(ev.dropped_attributes_count as i64),
         );
-        Value::Object(obj)
+        Value::Object(obj.into())
     }
 }
 
@@ -410,7 +410,7 @@ impl From<Link> for Value {
             "dropped_attributes_count".into(),
             Value::Integer(link.dropped_attributes_count as i64),
         );
-        Value::Object(obj)
+        Value::Object(obj.into())
     }
 }
 
@@ -419,6 +419,6 @@ impl From<SpanStatus> for Value {
         let mut obj: BTreeMap<KeyString, Value> = BTreeMap::new();
         obj.insert("message".into(), status.message.into());
         obj.insert("code".into(), status.code.into());
-        Value::Object(obj)
+        Value::Object(obj.into())
     }
 }

lib/vector-core/src/event/log_event.rs

@@ -7,7 +7,7 @@ use std::{
     num::NonZeroUsize,
     sync::{Arc, LazyLock},
 };
-
+use std::collections::BTreeMap;
 use crate::event::util::log::all_fields_skip_array_elements;
 use bytes::Bytes;
 use chrono::Utc;
@@ -150,7 +150,7 @@ impl PartialEq for Inner {
 #[derive(Clone, Debug, Default, Deserialize, PartialEq)]
 pub struct LogEvent {
     #[serde(flatten)]
-    inner: Arc<Inner>,
+    inner: Inner,
 
     #[serde(skip)]
     metadata: EventMetadata,
@@ -177,11 +177,11 @@ impl LogEvent {
     }
 
     pub fn value(&self) -> &Value {
-        self.inner.as_ref().as_value()
+        self.inner.as_value()
     }
 
     pub fn value_mut(&mut self) -> &mut Value {
-        let result = Arc::make_mut(&mut self.inner);
+        let result = &mut self.inner;
         // We MUST invalidate the inner size cache when making a
         // mutable copy, since the _next_ action will modify the data.
         result.invalidate();
@@ -258,24 +258,22 @@ impl LogEvent {
     ///  Create a `LogEvent` from a `Value` and `EventMetadata`
     pub fn from_parts(value: Value, metadata: EventMetadata) -> Self {
         Self {
-            inner: Arc::new(value.into()),
+            inner: value.into(),
             metadata,
         }
     }
 
     ///  Create a `LogEvent` from an `ObjectMap` and `EventMetadata`
-    pub fn from_map(map: ObjectMap, metadata: EventMetadata) -> Self {
-        let inner = Arc::new(Inner::from(Value::Object(map)));
+    pub fn from_map(map: BTreeMap<KeyString, Value>, metadata: EventMetadata) -> Self {
+        let inner = Inner::from(Value::Object(map.into()));
         Self { inner, metadata }
     }
 
     /// Convert a `LogEvent` into a tuple of its components
     pub fn into_parts(mut self) -> (Value, EventMetadata) {
         self.value_mut();
 
-        let value = Arc::try_unwrap(self.inner)
-            .unwrap_or_else(|_| unreachable!("inner fields already cloned after owning"))
-            .fields;
+        let value = self.inner.fields;
         let metadata = self.metadata;
         (value, metadata)
     }
@@ -668,6 +666,11 @@ impl From<ObjectMap> for LogEvent {
         Self::from_parts(Value::Object(map), EventMetadata::default())
     }
 }
+impl From<BTreeMap<KeyString, Value>> for LogEvent {
+    fn from(map: BTreeMap<KeyString, Value>) -> Self {
+        Self::from_parts(Value::Object(map.into()), EventMetadata::default())
+    }
+}
 
 impl From<HashMap<KeyString, Value>> for LogEvent {
     fn from(map: HashMap<KeyString, Value>) -> Self {

lib/vector-core/src/event/proto.rs

@@ -15,7 +15,7 @@ mod proto_event {
 pub use event_wrapper::Event;
 pub use metric::Value as MetricValue;
 pub use proto_event::*;
-use vrl::value::{ObjectMap, Value as VrlValue};
+use vrl::value::{Array, ObjectMap, Value as VrlValue};
 
 use super::{array, metric::MetricSketch, EventMetadata};
 
@@ -112,7 +112,7 @@ impl From<Log> for super::LogEvent {
                 .fields
                 .into_iter()
                 .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value)))
-                .collect::<ObjectMap>();
+                .collect();
 
             Self::from_map(fields, metadata)
         }
@@ -137,7 +137,7 @@ impl From<Trace> for super::TraceEvent {
             .fields
             .into_iter()
             .filter_map(|(k, v)| decode_value(v).map(|value| (k.into(), value)))
-            .collect::<ObjectMap>();
+            .collect();
 
         Self::from(super::LogEvent::from_map(fields, metadata))
     }
@@ -726,7 +726,7 @@ fn decode_array(items: Vec<Value>) -> Option<super::Value> {
     items
         .into_iter()
         .map(decode_value)
-        .collect::<Option<Vec<_>>>()
+        .collect::<Option<Array>>()
         .map(super::Value::Array)
 }
 
@@ -758,7 +758,7 @@ fn encode_map(fields: ObjectMap) -> ValueMap {
     }
 }
 
-fn encode_array(items: Vec<super::Value>) -> ValueArray {
+fn encode_array(items: Array) -> ValueArray {
     ValueArray {
         items: items.into_iter().map(encode_value).collect(),
     }

lib/vector-core/src/event/test/common.rs

@@ -1,8 +1,8 @@
 use std::{collections::BTreeSet, iter};
-
+use std::collections::BTreeMap;
 use chrono::{DateTime, Utc};
 use quickcheck::{empty_shrinker, Arbitrary, Gen};
-use vrl::value::{ObjectMap, Value};
+use vrl::value::Value;
 
 use super::super::{
     metric::{
@@ -77,7 +77,7 @@ impl Arbitrary for Event {
 impl Arbitrary for LogEvent {
     fn arbitrary(g: &mut Gen) -> Self {
         let mut gen = Gen::new(MAX_MAP_SIZE);
-        let map: ObjectMap = ObjectMap::arbitrary(&mut gen);
+        let map = BTreeMap::arbitrary(&mut gen);
         let metadata: EventMetadata = EventMetadata::arbitrary(g);
         LogEvent::from_map(map, metadata)
     }

lib/vector-core/src/event/trace.rs

@@ -30,7 +30,7 @@ impl TraceEvent {
     }
 
     pub fn from_parts(fields: ObjectMap, metadata: EventMetadata) -> Self {
-        Self(LogEvent::from_map(fields, metadata))
+        Self(LogEvent::from_parts(Value::Object(fields), metadata))
     }
 
     pub fn value(&self) -> &Value {

lib/vector-core/src/event/util/log/all_fields.rs

@@ -334,7 +334,7 @@ mod test {
             ("a.b.c", Value::Integer(5)),
             ("\"a.b.c\"", Value::Integer(6)),
             ("d", Value::Object(ObjectMap::new())),
-            ("e", Value::Array(Vec::new())),
+            ("e", Value::array()),
         ]
         .into_iter()
         .map(|(k, v)| (k.into(), v))
@@ -356,7 +356,7 @@ mod test {
             ("a.b.c", Value::Integer(5)),
             ("a.b.c", Value::Integer(6)),
             ("d", Value::Object(ObjectMap::new())),
-            ("e", Value::Array(Vec::new())),
+            ("e", Value::array()),
         ]
         .into_iter()
         .map(|(k, v)| (k.into(), v))