chore: Use value::Value directly in the codebase (vectordotdev#12673)

* Use `value::Value` directly in the codebase

This commit modifies the project to use `value::Value` directly and not as a
re-export from VRL. The changeset is large but almost entirely mechanical. I
would like to experiment with different representations of `Value` -- imagine a
world where the `Value` is mostly stack allocated, or where the `Value::Object`
doesn't recurse to another heap allocated `Value` outside of the current virtual
machine page -- but I realized while making some modifications to that type that
its representation was, well, kindly spread through the project.

We don't solve that here. Instead we take a single step in that
direction. Follow-up work will work a struct shim over the top of the `BTreeMap`
inside `Value` so callers are not aware of the `BTreeMap`. That abstraction is
what will allow us to fiddle with the representation, so long as we hew to the
enum nature of `Value`.

Signed-off-by: Brian L. Troutwine <[email protected]>

* test dings

Signed-off-by: Brian L. Troutwine <[email protected]>

* test dings

Signed-off-by: Brian L. Troutwine <[email protected]>

* reviews

Signed-off-by: Brian L. Troutwine <[email protected]>

* test dings

Signed-off-by: Brian L. Troutwine <[email protected]>

* newline ding

Signed-off-by: Brian L. Troutwine <[email protected]>

* clippy dings

Signed-off-by: Brian L. Troutwine <[email protected]>

Author: blt

.rustfmt.toml

@@ -1,7 +1,9 @@
-# Using defaults.
-# Not stable yet.
-# indent_style = "Block"
-# group_imports = "StdExternalCrate"
-# imports_granularity = "Crate"
-reorder_imports = true
 edition = "2021"
+newline_style = "unix"
+reorder_imports = true
+
+# Nightly only features
+# unstable_features = true
+# imports_granularity = "Crate"
+# group_imports = "StdExternalCrate"
+# indent_style = "Block"

Cargo.lock

@@ -113,7 +113,7 @@ lookup = { path = "lib/lookup" }
 portpicker = { path = "lib/portpicker" }
 prometheus-parser = { path = "lib/prometheus-parser", optional = true }
 tracing-limit = { path = "lib/tracing-limit" }
-value = { path = "lib/value", optional = true }
+value = { path = "lib/value" }
 vector_buffers = { path = "lib/vector-buffers", default-features = false }
 vector_common = { path = "lib/vector-common" }
 vector_core = { path = "lib/vector-core", default-features = false, features = ["vrl"] }
@@ -465,7 +465,7 @@ sources-aws_ecs_metrics = []
 sources-aws_kinesis_firehose = ["base64", "infer", "sources-utils-tls"]
 sources-aws_s3 = ["aws-core", "aws-sdk-sqs", "aws-sdk-s3", "semver", "async-compression", "sources-aws_sqs", "tokio-util/io"]
 sources-aws_sqs = ["aws-core", "aws-sdk-sqs"]
-sources-datadog_agent = ["sources-utils-tls", "sources-utils-http-error", "protobuf-build", "value"]
+sources-datadog_agent = ["sources-utils-tls", "sources-utils-http-error", "protobuf-build"]
 sources-demo_logs = ["fakedata"]
 sources-dnstap = ["base64", "trust-dns-proto", "dnsmsg-parser", "protobuf-build"]
 sources-docker_logs = ["docker"]
@@ -577,7 +577,7 @@ transforms-metric_to_log = []
 transforms-pipelines = ["transforms-filter", "transforms-route"]
 transforms-reduce = []
 transforms-regex_parser = []
-transforms-remap = ["value"]
+transforms-remap = []
 transforms-remove_fields = []
 transforms-remove_tags = []
 transforms-rename_fields = []

Cargo.toml

@@ -1,3 +1,5 @@
+use std::{fmt, time::Duration};
+
 use bytes::BytesMut;
 use codecs::{
     self,
@@ -8,7 +10,6 @@ use criterion::{
     criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
     SamplingMode, Throughput,
 };
-use std::{fmt, time::Duration};
 use tokio_util::codec::Decoder;
 
 #[derive(Debug)]

benches/codecs/character_delimited_bytes.rs

@@ -1,10 +1,11 @@
+use std::time::Duration;
+
 use bytes::{BufMut, BytesMut};
 use codecs::{encoding::Framer, JsonSerializer, NewlineDelimitedEncoder};
 use criterion::{
     criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, Criterion, SamplingMode,
     Throughput,
 };
-use std::time::Duration;
 use tokio_util::codec::Encoder;
 use vector::event::Event;
 use vector_common::{btreemap, byte_size_of::ByteSizeOf};

benches/codecs/encoder.rs

@@ -1,3 +1,5 @@
+use std::{fmt, time::Duration};
+
 use bytes::BytesMut;
 use codecs::{
     self, decoding::Deserializer, decoding::Framer, BytesDeserializer, NewlineDelimitedDecoder,
@@ -6,7 +8,6 @@ use criterion::{
     criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
     SamplingMode, Throughput,
 };
-use std::{fmt, time::Duration};
 use tokio_util::codec::Decoder;
 
 #[derive(Debug)]

benches/codecs/newline_bytes.rs

@@ -3,8 +3,8 @@ use std::{collections::BTreeMap, time::SystemTime};
 use chrono::prelude::*;
 use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
 use enrichment::Case;
+use value::Value;
 use vector::enrichment_tables::{file::File, Condition, Table};
-use vrl::Value;
 
 criterion_group!(
     name = benches;

benches/enrichment_tables_file.rs

@@ -74,9 +74,10 @@ impl Deserializer for BytesDeserializer {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use vector_core::config::log_schema;
 
+    use super::*;
+
     #[test]
     fn deserialize_bytes() {
         let input = Bytes::from("foo");

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

@@ -1,16 +1,18 @@
-use super::Deserializer;
+use std::convert::TryInto;
+
 use bytes::Bytes;
 use chrono::Utc;
 use serde::{Deserialize, Serialize};
 use smallvec::{smallvec, SmallVec};
-use std::convert::TryInto;
 use value::Kind;
 use vector_core::{
     config::{log_schema, DataType},
     event::Event,
     schema,
 };
 
+use super::Deserializer;
+
 /// Config used to build a `JsonDeserializer`.
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
 pub struct JsonDeserializerConfig;
@@ -100,9 +102,10 @@ impl From<&JsonDeserializerConfig> for JsonDeserializer {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use vector_core::config::log_schema;
 
+    use super::*;
+
     #[test]
     fn deserialize_json() {
         let input = Bytes::from(r#"{ "foo": 123 }"#);

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

@@ -10,19 +10,20 @@ mod native_json;
 #[cfg(feature = "syslog")]
 mod syslog;
 
-pub use self::bytes::{BytesDeserializer, BytesDeserializerConfig};
-#[cfg(feature = "syslog")]
-pub use self::syslog::{SyslogDeserializer, SyslogDeserializerConfig};
-pub use json::{JsonDeserializer, JsonDeserializerConfig};
-pub use native::{NativeDeserializer, NativeDeserializerConfig};
-pub use native_json::{NativeJsonDeserializer, NativeJsonDeserializerConfig};
+use std::fmt::Debug;
 
 use ::bytes::Bytes;
 use dyn_clone::DynClone;
+pub use json::{JsonDeserializer, JsonDeserializerConfig};
+pub use native::{NativeDeserializer, NativeDeserializerConfig};
+pub use native_json::{NativeJsonDeserializer, NativeJsonDeserializerConfig};
 use smallvec::SmallVec;
-use std::fmt::Debug;
 use vector_core::event::Event;
 
+pub use self::bytes::{BytesDeserializer, BytesDeserializerConfig};
+#[cfg(feature = "syslog")]
+pub use self::syslog::{SyslogDeserializer, SyslogDeserializerConfig};
+
 /// Parse structured events from bytes.
 pub trait Deserializer: DynClone + Debug + Send + Sync {
     /// Parses structured events from bytes.

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

@@ -2,9 +2,9 @@ use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use smallvec::{smallvec, SmallVec};
 use value::Kind;
+use vector_core::{config::DataType, event::Event, schema};
 
 use super::Deserializer;
-use vector_core::{config::DataType, event::Event, schema};
 
 /// Config used to build a `NativeJsonDeserializer`.
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
@@ -57,9 +57,10 @@ impl Deserializer for NativeJsonDeserializer {
 
 #[cfg(test)]
 mod test {
-    use super::*;
     use serde_json::json;
 
+    use super::*;
+
     #[test]
     fn parses_top_level_arrays() {
         let config = NativeJsonDeserializerConfig;

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

@@ -4,14 +4,14 @@ use serde::{Deserialize, Serialize};
 use smallvec::{smallvec, SmallVec};
 use syslog_loose::{IncompleteDate, Message, ProcId, Protocol};
 use value::Kind;
-
-use super::Deserializer;
 use vector_core::{
     config::{log_schema, DataType},
     event::{Event, Value},
     schema,
 };
 
+use super::Deserializer;
+
 /// Config used to build a `SyslogDeserializer`.
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
 pub struct SyslogDeserializerConfig;

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

@@ -148,10 +148,12 @@ impl Decoder for CharacterDelimitedDecoder {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use std::collections::HashMap;
+
     use bytes::BufMut;
     use indoc::indoc;
-    use std::collections::HashMap;
+
+    use super::*;
 
     #[test]
     fn decode() {

lib/codecs/src/decoding/framing/character_delimited.rs

@@ -9,23 +9,24 @@ mod length_delimited;
 mod newline_delimited;
 mod octet_counting;
 
-pub use self::bytes::{BytesDecoder, BytesDecoderConfig};
+use std::fmt::Debug;
+
+use ::bytes::Bytes;
 pub use character_delimited::{
     CharacterDelimitedDecoder, CharacterDelimitedDecoderConfig, CharacterDelimitedDecoderOptions,
 };
+use dyn_clone::DynClone;
 pub use length_delimited::{LengthDelimitedDecoder, LengthDelimitedDecoderConfig};
 pub use newline_delimited::{
     NewlineDelimitedDecoder, NewlineDelimitedDecoderConfig, NewlineDelimitedDecoderOptions,
 };
 pub use octet_counting::{
     OctetCountingDecoder, OctetCountingDecoderConfig, OctetCountingDecoderOptions,
 };
+use tokio_util::codec::LinesCodecError;
 
+pub use self::bytes::{BytesDecoder, BytesDecoderConfig};
 use super::StreamDecodingError;
-use ::bytes::Bytes;
-use dyn_clone::DynClone;
-use std::fmt::Debug;
-use tokio_util::codec::LinesCodecError;
 
 /// An error that occurred while producing byte frames from a byte stream / byte
 /// message.

lib/codecs/src/decoding/framing/mod.rs

@@ -5,6 +5,9 @@ mod error;
 pub mod format;
 pub mod framing;
 
+use std::fmt::Debug;
+
+use bytes::{Bytes, BytesMut};
 pub use error::StreamDecodingError;
 pub use format::{
     BoxedDeserializer, BytesDeserializer, BytesDeserializerConfig, JsonDeserializer,
@@ -20,11 +23,8 @@ pub use framing::{
     NewlineDelimitedDecoderConfig, NewlineDelimitedDecoderOptions, OctetCountingDecoder,
     OctetCountingDecoderConfig, OctetCountingDecoderOptions,
 };
-
-use bytes::{Bytes, BytesMut};
 use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
-use std::fmt::Debug;
 use vector_core::{config::DataType, event::Event, schema};
 
 /// An error that occurred while decoding structured events from a byte stream /

lib/codecs/src/decoding/mod.rs

@@ -58,11 +58,12 @@ impl Encoder<Event> for JsonSerializer {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use bytes::BytesMut;
     use vector_common::btreemap;
     use vector_core::event::Value;
 
+    use super::*;
+
     #[test]
     fn serialize_json() {
         let event = Event::from(btreemap! {

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

@@ -8,13 +8,13 @@ mod native;
 mod native_json;
 mod raw_message;
 
+use std::fmt::Debug;
+
+use dyn_clone::DynClone;
 pub use json::{JsonSerializer, JsonSerializerConfig};
 pub use native::{NativeSerializer, NativeSerializerConfig};
 pub use native_json::{NativeJsonSerializer, NativeJsonSerializerConfig};
 pub use raw_message::{RawMessageSerializer, RawMessageSerializerConfig};
-
-use dyn_clone::DynClone;
-use std::fmt::Debug;
 use vector_core::event::Event;
 
 /// Serialize a structured event into a byte frame.

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

@@ -1,14 +1,13 @@
+use bytes::{BufMut, BytesMut};
+use serde::{Deserialize, Serialize};
+use tokio_util::codec::Encoder;
+use value::Kind;
 use vector_core::{
     config::{log_schema, DataType},
     event::Event,
     schema,
 };
 
-use bytes::{BufMut, BytesMut};
-use serde::{Deserialize, Serialize};
-use tokio_util::codec::Encoder;
-use value::Kind;
-
 /// Config used to build a `RawMessageSerializer`.
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
 pub struct RawMessageSerializerConfig;
@@ -71,9 +70,10 @@ impl Encoder<Event> for RawMessageSerializer {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use bytes::{Bytes, BytesMut};
 
+    use super::*;
+
     #[test]
     fn serialize_bytes() {
         let input = Event::from("foo");

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

@@ -8,17 +8,18 @@ mod character_delimited;
 mod length_delimited;
 mod newline_delimited;
 
-pub use self::bytes::{BytesEncoder, BytesEncoderConfig};
+use std::fmt::Debug;
+
 pub use character_delimited::{
     CharacterDelimitedEncoder, CharacterDelimitedEncoderConfig, CharacterDelimitedEncoderOptions,
 };
+use dyn_clone::DynClone;
 pub use length_delimited::{LengthDelimitedEncoder, LengthDelimitedEncoderConfig};
 pub use newline_delimited::{NewlineDelimitedEncoder, NewlineDelimitedEncoderConfig};
-
-use dyn_clone::DynClone;
-use std::fmt::Debug;
 use tokio_util::codec::LinesCodecError;
 
+pub use self::bytes::{BytesEncoder, BytesEncoderConfig};
+
 /// An error that occurred while framing bytes.
 pub trait FramingError: std::error::Error + Send + Sync {}