dataflow_rs/base/
context.rs

1//! The context, `{data, metadata, temp_data}`, on v3's wire shape, and the
2//! reader every value of a wire shape is read back with.
3//!
4//! The context's reads and writes over a [`Path`](crate::expr::Path) live in
5//! `run`, beside the message, since `expr` sits above `base`.
6
7use datavalue::{NumberValue, OwnedDataValue};
8use serde::de::{self, MapAccess, SeqAccess, Visitor};
9use serde::{Deserialize, Deserializer, Serialize, Serializer};
10use std::collections::HashSet;
11use std::fmt;
12
13/// The three roots, in the order a context holds them.
14pub(crate) const ROOTS: [&str; 3] = ["data", "metadata", "temp_data"];
15
16/// The context: always `data`, `metadata` and `temp_data`, in that order.
17/// Each root starts as an object, and every write keeps it one: `map`,
18/// `Ctx::set` and [`Context::set`] refuse anything else at a root.
19///
20/// Read from JSON, it takes the three roots in any order and writes them in
21/// that order, reads a missing one as `{}`, takes any value at each (a
22/// trace's redacted root is `null`), and refuses any other key,
23/// since expressions evaluate against the whole object and a fourth key
24/// would be readable by `{"var": …}`. An object anywhere in it holding a key
25/// twice is refused too: the engine reads the first, where a JSON parser
26/// keeps the last, so the two would disagree on what the context says.
27#[derive(Clone, Debug, PartialEq)]
28pub struct Context {
29    pub(crate) root: OwnedDataValue,
30}
31
32/// A seed for a root: an object as it is, anything else `{}` with capacity
33/// `cap`.
34fn slot(seed: OwnedDataValue, cap: usize) -> OwnedDataValue {
35    if seed.is_object() {
36        seed
37    } else {
38        OwnedDataValue::Object(Vec::with_capacity(cap))
39    }
40}
41
42impl Context {
43    /// A context seeded with `data` and `metadata`, each ignored when it is
44    /// not an object, and an empty `temp_data`. `metadata` is sized for the
45    /// engine's run stamps and progress object so the first run does not
46    /// reallocate it.
47    pub(crate) fn new(data: OwnedDataValue, metadata: OwnedDataValue) -> Self {
48        Self::from_roots([
49            slot(data, 0),
50            slot(metadata, 4),
51            OwnedDataValue::Object(Vec::new()),
52        ])
53    }
54
55    /// A context over `root`, the object holding the three roots: a trace's
56    /// `initial` and each context its replay rebuilds.
57    pub(crate) fn from_root(root: OwnedDataValue) -> Self {
58        Self { root }
59    }
60
61    /// A context over the three roots' values, in [`ROOTS`]' order. A
62    /// literal, not an iterator over `ROOTS`: every message is built through
63    /// it, and the zipped iterator cost about 0.5% of `micro_cond_bench`'s
64    /// instructions.
65    fn from_roots([data, metadata, temp_data]: [OwnedDataValue; 3]) -> Self {
66        let [d, m, t] = ROOTS;
67        Self::from_root(OwnedDataValue::Object(vec![
68            (d.to_owned(), data),
69            (m.to_owned(), metadata),
70            (t.to_owned(), temp_data),
71        ]))
72    }
73
74    fn root(&self, key: &str) -> &OwnedDataValue {
75        &self.root[key]
76    }
77
78    /// The `data` root.
79    pub fn data(&self) -> &OwnedDataValue {
80        self.root("data")
81    }
82
83    /// The `metadata` root.
84    pub fn metadata(&self) -> &OwnedDataValue {
85        self.root("metadata")
86    }
87
88    /// The `temp_data` root.
89    pub fn temp_data(&self) -> &OwnedDataValue {
90        self.root("temp_data")
91    }
92
93    /// The object holding the three roots: what expressions evaluate against.
94    pub fn as_value(&self) -> &OwnedDataValue {
95        &self.root
96    }
97}
98
99impl Serialize for Context {
100    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
101        self.root.serialize(s)
102    }
103}
104
105impl<'de> Deserialize<'de> for Context {
106    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
107        d.deserialize_map(Roots)
108    }
109}
110
111#[cfg(feature = "ts")]
112crate::base::ts::ts_decl!(
113    Context,
114    "Context",
115    [
116        " The context: `data`, `metadata` and `temp_data`, in that order. Each",
117        " root starts as an object; a trace's redacted root is `null`.",
118    ],
119    |_cfg| {
120        use crate::base::ts::{Prop, object};
121        object(&[
122            Prop::new("data", "JsonValue", &[" The `data` root."]),
123            Prop::new("metadata", "JsonValue", &[" The `metadata` root."]),
124            Prop::new("temp_data", "JsonValue", &[" The `temp_data` root."]),
125        ])
126    }
127);
128
129/// [`Context`]'s reader: the three roots, by name, once each.
130struct Roots;
131
132impl<'de> Visitor<'de> for Roots {
133    type Value = Context;
134
135    fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        f.write_str("a context: an object holding `data`, `metadata` and `temp_data`")
137    }
138
139    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Context, A::Error> {
140        let mut values: [Option<OwnedDataValue>; 3] = [None, None, None];
141        while let Some(key) = map.next_key::<String>()? {
142            let Some(at) = ROOTS.iter().position(|root| *root == key) else {
143                return Err(de::Error::custom(format_args!(
144                    "a context holds only `data`, `metadata` and `temp_data`, not `{key}`"
145                )));
146            };
147            if values[at].is_some() {
148                return Err(de::Error::duplicate_field(ROOTS[at]));
149            }
150            values[at] = Some(map.next_value::<ReadValue<false>>()?.0);
151        }
152        Ok(Context::from_roots(values.map(|value| {
153            value.unwrap_or_else(|| OwnedDataValue::Object(Vec::new()))
154        })))
155    }
156}
157
158/// A value read back as v4 wrote it: as [`OwnedDataValue`]'s own reader
159/// reads it, except that a number written with a fraction or an exponent
160/// stays a float, as `OwnedDataValue::from(&serde_json::Value)` keeps one,
161/// and an object holding a key twice is refused. That reader makes `2.0` the
162/// integer `2`, which writes back as `2`, so a message holding a
163/// whole-number float would not read back byte for byte; and it keeps both
164/// pairs of a repeated key, which every path rule reads as one.
165pub(crate) fn exact<'de, D: Deserializer<'de>>(d: D) -> Result<OwnedDataValue, D::Error> {
166    d.deserialize_any(ValueVisitor::<false>)
167}
168
169/// A value read in document order, folding any repeated object key to its
170/// first position and last value (as `serde_json::Value` with `preserve_order`
171/// reads one).
172pub(crate) fn in_order<'de, D: Deserializer<'de>>(d: D) -> Result<OwnedDataValue, D::Error> {
173    d.deserialize_any(ValueVisitor::<true>)
174}
175
176/// A value read by [`ValueVisitor`].
177struct ReadValue<const DEDUP: bool>(OwnedDataValue);
178
179impl<'de, const DEDUP: bool> Deserialize<'de> for ReadValue<DEDUP> {
180    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
181        d.deserialize_any(ValueVisitor::<DEDUP>).map(Self)
182    }
183}
184
185/// [`exact`]'s and [`in_order`]'s reader: `OwnedDataValue`'s, keeping floats
186/// as floats and either refusing (`DEDUP = false`) or folding (`DEDUP = true`)
187/// repeated object keys.
188struct ValueVisitor<const DEDUP: bool>;
189
190impl<'de, const DEDUP: bool> Visitor<'de> for ValueVisitor<DEDUP> {
191    type Value = OwnedDataValue;
192
193    fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
194        f.write_str("any valid JSON value")
195    }
196
197    fn visit_unit<E: de::Error>(self) -> Result<OwnedDataValue, E> {
198        Ok(OwnedDataValue::Null)
199    }
200
201    fn visit_none<E: de::Error>(self) -> Result<OwnedDataValue, E> {
202        Ok(OwnedDataValue::Null)
203    }
204
205    fn visit_some<D: Deserializer<'de>>(self, d: D) -> Result<OwnedDataValue, D::Error> {
206        d.deserialize_any(Self)
207    }
208
209    fn visit_bool<E: de::Error>(self, v: bool) -> Result<OwnedDataValue, E> {
210        Ok(OwnedDataValue::Bool(v))
211    }
212
213    fn visit_i64<E: de::Error>(self, v: i64) -> Result<OwnedDataValue, E> {
214        Ok(OwnedDataValue::Number(NumberValue::Integer(v)))
215    }
216
217    fn visit_u64<E: de::Error>(self, v: u64) -> Result<OwnedDataValue, E> {
218        Ok(OwnedDataValue::Number(NumberValue::from_u64(v)))
219    }
220
221    fn visit_i128<E: de::Error>(self, v: i128) -> Result<OwnedDataValue, E> {
222        Ok(OwnedDataValue::Number(match i64::try_from(v) {
223            Ok(i) => NumberValue::Integer(i),
224            Err(_) => NumberValue::Float(v as f64),
225        }))
226    }
227
228    fn visit_u128<E: de::Error>(self, v: u128) -> Result<OwnedDataValue, E> {
229        Ok(OwnedDataValue::Number(match i64::try_from(v) {
230            Ok(i) => NumberValue::Integer(i),
231            Err(_) => NumberValue::Float(v as f64),
232        }))
233    }
234
235    fn visit_f64<E: de::Error>(self, v: f64) -> Result<OwnedDataValue, E> {
236        Ok(if v.is_finite() {
237            OwnedDataValue::Number(NumberValue::Float(v))
238        } else {
239            OwnedDataValue::Null
240        })
241    }
242
243    fn visit_str<E: de::Error>(self, v: &str) -> Result<OwnedDataValue, E> {
244        Ok(OwnedDataValue::String(v.to_owned()))
245    }
246
247    fn visit_string<E: de::Error>(self, v: String) -> Result<OwnedDataValue, E> {
248        Ok(OwnedDataValue::String(v))
249    }
250
251    fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<OwnedDataValue, A::Error> {
252        let mut items = Vec::with_capacity(seq.size_hint().unwrap_or(0));
253        while let Some(ReadValue(item)) = seq.next_element::<ReadValue<DEDUP>>()? {
254            items.push(item);
255        }
256        Ok(OwnedDataValue::Array(items))
257    }
258
259    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<OwnedDataValue, A::Error> {
260        let mut pairs = Vec::with_capacity(map.size_hint().unwrap_or(0));
261        while let Some((key, ReadValue(value))) = map.next_entry::<String, ReadValue<DEDUP>>()? {
262            pairs.push((key, value));
263        }
264        if DEDUP {
265            dedup_object(&mut pairs);
266        } else if let Some(key) = repeated_key(&pairs) {
267            return Err(de::Error::custom(format_args!(
268                "an object holds the key `{key}` twice"
269            )));
270        }
271        Ok(OwnedDataValue::Object(pairs))
272    }
273}
274
275/// Above this many keys, an object finds a repeated key through a set
276/// rather than a scan.
277const SCAN: usize = 8;
278
279/// The first key `pairs` holds a second time, if any.
280fn repeated_key(pairs: &[(String, OwnedDataValue)]) -> Option<&str> {
281    if pairs.len() <= SCAN {
282        return pairs
283            .iter()
284            .enumerate()
285            .find(|(i, (key, _))| pairs[..*i].iter().any(|(k, _)| k == key))
286            .map(|(_, (key, _))| key.as_str());
287    }
288    let mut seen = HashSet::with_capacity(pairs.len());
289    pairs
290        .iter()
291        .map(|(key, _)| key.as_str())
292        .find(|key| !seen.insert(*key))
293}
294
295/// Ensures `pairs` holds each key once: a repeated key keeps its first
296/// position and its last value, as `serde_json::Value` reads one.
297fn dedup_object(pairs: &mut Vec<(String, OwnedDataValue)>) {
298    if repeated_key(pairs).is_some() {
299        fold_repeats(pairs);
300    }
301}
302
303/// `value` with every object in it holding each key once: a repeated key
304/// keeps its first position and its last value, as `serde_json::Value`
305/// reads one.
306pub(crate) fn keep_last(value: &mut OwnedDataValue) {
307    match value {
308        OwnedDataValue::Object(pairs) => {
309            dedup_object(pairs);
310            pairs.iter_mut().for_each(|(_, v)| keep_last(v));
311        }
312        OwnedDataValue::Array(items) => items.iter_mut().for_each(keep_last),
313        _ => {}
314    }
315}
316
317/// [`keep_last`] on one object known to hold a key twice.
318#[cold]
319fn fold_repeats(pairs: &mut Vec<(String, OwnedDataValue)>) {
320    let mut len = 0;
321    for i in 0..pairs.len() {
322        let (key, value) = std::mem::take(&mut pairs[i]);
323        if let Some((_, slot)) = pairs[..len].iter_mut().find(|(k, _)| *k == key) {
324            *slot = value;
325        } else {
326            pairs[len] = (key, value);
327            len += 1;
328        }
329    }
330    pairs.truncate(len);
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336    use serde_json::json;
337
338    fn text(c: &Context) -> String {
339        serde_json::to_string(c).unwrap()
340    }
341
342    #[test]
343    fn a_seed_that_is_not_an_object_is_ignored() {
344        let c = Context::new(
345            OwnedDataValue::from(&json!([1])),
346            OwnedDataValue::from(&json!({"m": 1})),
347        );
348        assert_eq!(text(&c), r#"{"data":{},"metadata":{"m":1},"temp_data":{}}"#);
349    }
350
351    #[test]
352    fn a_context_reads_what_it_writes() {
353        let wire = r#"{"data":{"b":1,"a":[null]},"metadata":null,"temp_data":"t"}"#;
354        let c: Context = serde_json::from_str(wire).unwrap();
355        assert_eq!(text(&c), wire);
356        assert_eq!(c.metadata(), &OwnedDataValue::Null);
357        assert_eq!(
358            c,
359            Context::from_root(OwnedDataValue::from(&json!({
360                "data": {"b": 1, "a": [null]}, "metadata": null, "temp_data": "t"
361            })))
362        );
363    }
364
365    #[test]
366    fn a_value_reads_back_as_written_and_as_the_json_bridge_builds_it() {
367        let wire =
368            r#"[2.0,-3.0,2.5,1e+300,7,-7,9223372036854775807,"s",true,null,{"b":[1.0],"a":{}}]"#;
369        let mut de = serde_json::Deserializer::from_str(wire);
370        let read = exact(&mut de).unwrap();
371        assert_eq!(serde_json::to_string(&read).unwrap(), wire);
372        let bridged: serde_json::Value = serde_json::from_str(wire).unwrap();
373        assert_eq!(read, OwnedDataValue::from(&bridged));
374        // `OwnedDataValue`'s own reader makes the whole floats integers.
375        let collapsed: OwnedDataValue = serde_json::from_str(wire).unwrap();
376        assert!(
377            serde_json::to_string(&collapsed)
378                .unwrap()
379                .starts_with("[2,-3,2.5,")
380        );
381        // Above `i64::MAX`, an integer is a float, as the bridge makes it.
382        let big = exact(&mut serde_json::Deserializer::from_str(
383            "18446744073709551615",
384        ))
385        .unwrap();
386        assert_eq!(big, OwnedDataValue::from(&json!(u64::MAX)));
387    }
388    #[test]
389    fn a_repeated_key_keeps_its_first_position_and_its_last_value() {
390        let fold = |text: &str| {
391            let mut value = OwnedDataValue::from_json(text).unwrap();
392            keep_last(&mut value);
393            serde_json::to_string(&value).unwrap()
394        };
395        assert_eq!(
396            fold(r#"{"a":1,"b":2,"a":3,"c":4,"b":5}"#),
397            r#"{"a":3,"b":5,"c":4}"#
398        );
399        // Inside arrays and objects, and inside a repeated key's last value.
400        assert_eq!(
401            fold(r#"[{"x":{"y":1,"y":2}},{"a":{"q":1},"a":{"q":2,"q":3}}]"#),
402            r#"[{"x":{"y":2}},{"a":{"q":3}}]"#
403        );
404        // Past the scan, through the set.
405        let wide: String = (0..12).map(|i| format!(r#""k{i}":{i},"#)).collect();
406        let wanted: String = (0..12)
407            .map(|i| format!(r#""k{i}":{},"#, if i == 3 { 99 } else { i }))
408            .collect();
409        assert_eq!(
410            fold(&format!(r#"{{{wide}"k3":99}}"#)),
411            format!("{{{}}}", wanted.trim_end_matches(','))
412        );
413        // Nothing repeated, nothing moves.
414        let plain = r#"{"b":1,"a":{"b":2},"c":[{"b":3}]}"#;
415        assert_eq!(fold(plain), plain);
416    }
417}
dataflow_rs/base/context.rs

dataflow_rs/base/
context.rs
