dataflow_rs/expr/
config.rs

1//! Config fields: the types a function's `input` is made of, and the context
2//! that compiles them with every issue at its own path.
3
4use super::document::Interpolated;
5use super::engine::ExprEngine;
6use super::expr::{Expr, FromValue, ResolveExpr};
7use crate::base::typo::unknown_key;
8use crate::base::{AuthoredPath, ErrorKind, Issue, IssueCode, PathSeg, Result, Severity, StepKind};
9use datavalue::{NumberValue, OwnedDataValue};
10use serde_json::Value;
11use std::borrow::Cow;
12use std::collections::BTreeMap;
13use std::fmt;
14
15/// What a field's value is, for a field's schema and for checking a literal.
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17#[non_exhaustive]
18pub enum ValueKind {
19    /// Any JSON.
20    Any,
21    /// A string.
22    String,
23    /// A number.
24    Number,
25    /// A boolean.
26    Bool,
27    /// An object.
28    Object,
29    /// An array.
30    Array,
31    /// A context path, rooted at `data`, `metadata` or `temp_data`.
32    Path,
33}
34
35/// What a config field is, for tooling.
36#[derive(Clone, Copy, Debug)]
37#[non_exhaustive]
38pub enum FieldKind {
39    /// A literal of the given kind, never evaluated.
40    Literal(ValueKind),
41    /// JSONLogic resolving to the given kind.
42    Expr(ValueKind),
43    /// A document with expression holes.
44    Interpolated,
45    /// A write destination.
46    Output,
47    /// A runtime-declared set of fields.
48    Templated,
49    /// An array of the element kind.
50    List(fn() -> Self),
51    /// An object whose values are of the given kind.
52    Map(fn() -> Self),
53    /// A nested config's fields.
54    Nested(fn() -> &'static [FieldSpec]),
55}
56
57impl FieldKind {
58    /// Whether the kind is an output, or holds one at any depth of a list or
59    /// a map. The engine records every output's result whatever its field
60    /// says, so a field of such a kind records. A nested config's
61    /// own fields say so for themselves.
62    pub(crate) fn contains_output(self) -> bool {
63        match self {
64            Self::Output => true,
65            Self::List(element) | Self::Map(element) => element().contains_output(),
66            _ => false,
67        }
68    }
69}
70
71/// One field of a config.
72#[derive(Clone, Debug)]
73#[non_exhaustive]
74pub struct FieldSpec {
75    /// The authored key.
76    pub name: Cow<'static, str>,
77    /// What the field is.
78    pub kind: FieldKind,
79    /// Is the key required?
80    pub required: bool,
81    /// Does the engine record the field's result? Always for a field holding
82    /// an output.
83    pub records: bool,
84    /// The default, as JSON text.
85    pub default: Option<Cow<'static, str>>,
86    /// Is this the primary output?
87    pub primary: bool,
88    /// The field's doc comment.
89    pub description: Option<Cow<'static, str>>,
90}
91
92impl FieldSpec {
93    /// A runtime field: its kind is an expression, an interpolated document, a
94    /// literal or an output. It records when it holds an output.
95    pub fn runtime(name: impl Into<Cow<'static, str>>, kind: FieldKind, required: bool) -> Self {
96        Self {
97            name: name.into(),
98            kind,
99            required,
100            records: kind.contains_output(),
101            default: None,
102            primary: false,
103            description: None,
104        }
105    }
106
107    /// Sets the default, as JSON text.
108    #[must_use]
109    pub fn with_default(mut self, json: impl Into<Cow<'static, str>>) -> Self {
110        self.default = Some(json.into());
111        self
112    }
113}
114
115/// Receives every expression of a config, with its path.
116pub trait ExprVisitor {
117    /// An expression at `at`; `records` is set when the engine records its
118    /// result.
119    fn expr(&mut self, _at: &AuthoredPath, _raw: &Value, _records: bool) {}
120
121    /// An interpolated document at `at`.
122    fn interpolated(&mut self, _at: &AuthoredPath, _doc: &Interpolated) {}
123}
124
125/// A walk over a compiled config's expressions.
126pub struct Visit<'v> {
127    visitor: &'v mut dyn ExprVisitor,
128    path: AuthoredPath,
129    records: bool,
130}
131
132impl<'v> Visit<'v> {
133    /// A walk reporting to `visitor`, with paths below `root`.
134    pub fn new(visitor: &'v mut dyn ExprVisitor, root: AuthoredPath) -> Self {
135        Self {
136            visitor,
137            path: root,
138            records: false,
139        }
140    }
141
142    /// Runs `f` one segment deeper.
143    pub fn at(&mut self, seg: impl Into<PathSeg>, f: impl FnOnce(&mut Self)) {
144        self.path.push(seg);
145        f(self);
146        self.path.pop();
147    }
148
149    /// Runs `f` with the `records` flag ORed in, restoring it after.
150    pub fn recording(&mut self, records: bool, f: impl FnOnce(&mut Self)) {
151        let before = self.records;
152        self.records |= records;
153        f(self);
154        self.records = before;
155    }
156
157    /// Reports an expression at the current path.
158    pub fn expr(&mut self, raw: &Value) {
159        self.visitor.expr(&self.path, raw, self.records);
160    }
161
162    /// Reports an interpolated document at the current path.
163    pub fn interpolated(&mut self, doc: &Interpolated) {
164        self.visitor.interpolated(&self.path, doc);
165    }
166}
167
168impl fmt::Debug for Visit<'_> {
169    /// The current path and `records` flag; the visitor is not shown.
170    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
171        f.debug_struct("Visit")
172            .field("path", &format_args!("{}", self.path))
173            .field("records", &self.records)
174            .finish_non_exhaustive()
175    }
176}
177
178/// Compiles config fields, recording each issue at its exact authored path.
179pub struct CompileCx<'a> {
180    exprs: &'a ExprEngine,
181    issues: &'a mut Vec<Issue>,
182    path: AuthoredPath,
183    root_depth: usize,
184    step: Option<(StepKind, String)>,
185    runtime_fields: Option<&'a [FieldSpec]>,
186    rejected: usize,
187}
188
189impl<'a> CompileCx<'a> {
190    /// A context rooted at the document itself.
191    pub fn new(exprs: &'a ExprEngine, issues: &'a mut Vec<Issue>) -> Self {
192        Self {
193            exprs,
194            issues,
195            path: AuthoredPath::root(),
196            root_depth: 0,
197            step: None,
198            runtime_fields: None,
199            rejected: 0,
200        }
201    }
202
203    /// Roots the walk at `path`; a field is the first key below it.
204    #[must_use]
205    pub fn at_path(mut self, path: AuthoredPath) -> Self {
206        self.root_depth = path.segs().len();
207        self.path = path;
208        self
209    }
210
211    /// Stamps the step on every issue.
212    #[must_use]
213    pub fn for_step(mut self, kind: StepKind, id: impl Into<String>) -> Self {
214        self.step = Some((kind, id.into()));
215        self
216    }
217
218    /// The fields a plugin declared at runtime.
219    #[must_use]
220    pub fn with_runtime_fields(mut self, fields: &'a [FieldSpec]) -> Self {
221        self.runtime_fields = Some(fields);
222        self
223    }
224
225    /// The engine expressions compile with.
226    pub fn exprs(&self) -> &ExprEngine {
227        self.exprs
228    }
229
230    /// The current path.
231    pub fn path(&self) -> &AuthoredPath {
232        &self.path
233    }
234
235    /// The runtime-declared fields, if any.
236    pub fn runtime_fields(&self) -> Option<&'a [FieldSpec]> {
237        self.runtime_fields
238    }
239
240    /// Records an issue at the current path, at the code's default severity.
241    pub fn issue(&mut self, code: IssueCode, message: impl Into<String>) {
242        self.issue_as(code, code.severity(), message);
243    }
244
245    /// Records an issue at the current path with an explicit severity.
246    pub fn issue_as(&mut self, code: IssueCode, severity: Severity, message: impl Into<String>) {
247        let issue = Issue::new(code, self.path.clone(), message).with_severity(severity);
248        self.record(issue);
249    }
250
251    /// Records an issue found by another walk, such as a function's own
252    /// checks, whose path is relative to the current path. The path is
253    /// prefixed with the current one, the step and the field are stamped as
254    /// [`Self::issue`] stamps them, and a Rejected issue is counted.
255    pub(crate) fn adopt(&mut self, mut issue: Issue) {
256        issue.path = self.path.extend(&issue.path);
257        self.record(issue);
258    }
259
260    /// Stamps the step and the field on `issue`, counts it when it is
261    /// Rejected, and records it.
262    fn record(&mut self, mut issue: Issue) {
263        if let Some((kind, id)) = &self.step {
264            issue = issue.with_step(*kind, Some(id));
265        }
266        if let Some(PathSeg::Key(key)) = issue.path.segs().get(self.root_depth) {
267            issue.field = Some(key.to_string());
268        }
269        if issue.severity == Severity::Rejected {
270            self.rejected += 1;
271        }
272        self.issues.push(issue);
273    }
274
275    /// Runs `f` one segment deeper.
276    pub fn at<R>(&mut self, seg: impl Into<PathSeg>, f: impl FnOnce(&mut Self) -> R) -> R {
277        self.path.push(seg);
278        let out = f(self);
279        self.path.pop();
280        out
281    }
282
283    /// How many Rejected issues this context has recorded.
284    pub fn rejected(&self) -> usize {
285        self.rejected
286    }
287
288    /// Records an unknown `key` at the current path (the caller is inside
289    /// `at(key, …)`): a Rejected `LIKELY_TYPO` naming the near key, else a
290    /// Rejected `UNKNOWN_KEY`.
291    #[doc(hidden)]
292    pub fn unknown_key(&mut self, key: &str, known: &[&str]) {
293        let (code, message) = unknown_key(key, known);
294        self.issue_as(code, Severity::Rejected, message);
295    }
296}
297
298impl fmt::Debug for CompileCx<'_> {
299    /// The current path, the step, and how many issues are recorded and
300    /// Rejected; not the issues or the engine.
301    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
302        f.debug_struct("CompileCx")
303            .field("path", &format_args!("{}", self.path))
304            .field("step", &self.step)
305            .field("issues", &self.issues.len())
306            .field("rejected", &self.rejected)
307            .finish_non_exhaustive()
308    }
309}
310
311/// What [`Resolver`] is sealed with: only the crate implements it.
312pub(crate) mod sealed {
313    /// The seal: a type the crate resolves configs with.
314    pub trait Sealed {}
315}
316
317/// What resolving a config needs from a call: an expression evaluated, and
318/// a document rendered, each to an owned value, against the context the call
319/// sees. A [`ConfigField`] calls it from [`ConfigField::resolve`]; a call's
320/// context implements it ([`Ctx`](crate::Ctx), through a shared reference).
321///
322/// Sealed: a host calls it, from its own [`ConfigField::resolve`], and never
323/// implements it, so a method can join it without a breaking change.
324///
325/// ```compile_fail
326/// use dataflow_rs::Result;
327/// use dataflow_rs::datavalue::OwnedDataValue;
328/// use dataflow_rs::expr::{Interpolated, ResolveExpr, Resolver};
329///
330/// struct Mine;
331///
332/// impl Resolver for Mine {
333///     fn eval(&mut self, _expr: &dyn ResolveExpr) -> Result<OwnedDataValue> {
334///         Ok(OwnedDataValue::Null)
335///     }
336///     fn render(&mut self, _doc: &Interpolated) -> Result<OwnedDataValue> {
337///         Ok(OwnedDataValue::Null)
338///     }
339/// }
340/// ```
341pub trait Resolver: sealed::Sealed {
342    /// `expr` evaluated, as an owned value: a number for a `u64`, `i64` or
343    /// `f64` expression, a boolean for a `bool` one, a string for a `String`
344    /// one, a path's dotted text for a `Path` one, and the value itself for
345    /// any other. A folded constant is its value, evaluated never.
346    ///
347    /// # Errors
348    ///
349    /// The evaluation's: [`ErrorKind::Budget`] when the operation ceiling is
350    /// exceeded, [`ErrorKind::Eval`] for any other failure, a result that is
351    /// not of the expression's type included.
352    fn eval(&mut self, expr: &dyn ResolveExpr) -> Result<OwnedDataValue>;
353
354    /// `doc` rendered, as an owned tree: each hole's value, or its default.
355    ///
356    /// # Errors
357    ///
358    /// None from a call's context, which is every resolver there is today;
359    /// the `Result` leaves room for one that may refuse.
360    fn render(&mut self, doc: &Interpolated) -> Result<OwnedDataValue>;
361}
362
363/// A type a config is made of.
364///
365/// An `Option<T>` field is optional: an absent key is `None`, and so is an
366/// explicit `null`, unless `T` takes `null` as a value
367/// ([`NULL_IS_VALUE`](Self::NULL_IS_VALUE)), as an [`Expr`], an
368/// [`Interpolated`] document and any JSON value do. There the `null` is
369/// compiled as `T`: `map`'s `"logic": null` is a constant `null` expression,
370/// not a missing `logic`.
371///
372/// A host writes its own field type by implementing it: here a connector's
373/// name, a newtype over an `Expr<String>`, compiled, walked and resolved as
374/// the expression is, so a derived config holding one checks its secrets
375/// and template keys, and resolves it, as it would the expression.
376///
377/// ```
378/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
379/// use dataflow_rs::datavalue::OwnedDataValue;
380/// use dataflow_rs::expr::{CompileCx, ConfigField, FieldKind, Resolver, ValueKind, Visit};
381/// use dataflow_rs::function::{FunctionConfig, testing};
382/// use dataflow_rs::{Ctx, Expr, Function, Message, Outcome, Result};
383/// use serde_json::{Value, json};
384///
385/// /// The connector a call reaches, by name.
386/// struct Connector(Expr<String>);
387///
388/// impl ConfigField for Connector {
389///     fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
390///         Expr::<String>::compile(raw, cx).map(Self)
391///     }
392///     fn visit(&self, v: &mut Visit<'_>) {
393///         v.expr(self.0.raw());
394///     }
395///     fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
396///         r.eval(&self.0).map(Some)
397///     }
398///     fn kind() -> FieldKind {
399///         FieldKind::Expr(ValueKind::String)
400///     }
401/// }
402///
403/// #[derive(FunctionConfig)]
404/// struct Query {
405///     connector: Connector,
406///     limit: Option<Expr<u64>>,
407/// }
408///
409/// /// Writes its input, as a layer or a dry run reads it, at `data.input`.
410/// struct Echo;
411///
412/// impl Function for Echo {
413///     type Config = Query;
414///
415///     async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &Query) -> Result<Outcome> {
416///         let input = ctx.input()?;
417///         ctx.set(&"data.input".parse()?, input)?;
418///         Ok(Outcome::Ok)
419///     }
420/// }
421///
422/// let input = json!({"connector": {"cat": ["db-", {"var": "data.region"}]}});
423/// let data = OwnedDataValue::from(&json!({"region": "eu"}));
424/// let out = testing::call(Echo, input)
425///     .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
426///     .await;
427/// assert_eq!(out.outcome.unwrap(), Outcome::Ok);
428/// // The field resolved at its key; the `Option` left unset is absent.
429/// let at = "data.input".parse().unwrap();
430/// let written = out.message.context().get(&at).unwrap();
431/// assert_eq!(serde_json::to_string(written).unwrap(), r#"{"connector":"db-eu"}"#);
432/// # });
433/// ```
434pub trait ConfigField: Sized + Send + Sync + 'static {
435    /// Whether `null` is one of the type's values. When it is not, an explicit
436    /// `null` in an `Option<Self>` field means `None`.
437    const NULL_IS_VALUE: bool = false;
438
439    /// Whether the field is an output: a destination its call writes to.
440    /// `false` unless the type is [`Output`](crate::expr::Output) or an
441    /// `Option` of one. The derive reads a field typed `Output` or marked
442    /// `#[output]` as an output and asserts that this is `true`, so a type
443    /// that merely shares the name is refused at compile time.
444    const IS_OUTPUT: bool = false;
445
446    /// Compiles `raw`; `None` after recording at least one issue.
447    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self>;
448
449    /// The value when the key is absent and no default is declared. `None`
450    /// makes the field required.
451    fn absent() -> Option<Self> {
452        None
453    }
454
455    /// Reports the field's expressions.
456    fn visit(&self, _v: &mut Visit<'_>) {}
457
458    /// The field's value for this call, resolved through `r`, or `None` when
459    /// it has none (an `Option` the input left unset, with no default). The
460    /// value sits where [`Self::visit`] reports the field's expressions: a
461    /// list's elements at their indices, a map's values at their keys, and
462    /// an element left unset is `null`, keeping its place.
463    ///
464    /// # Errors
465    ///
466    /// The resolver's, for the first expression or document that fails.
467    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>>;
468
469    /// What the field is.
470    fn kind() -> FieldKind;
471}
472
473/// A scalar's value: itself.
474fn owned<T: Clone + Into<OwnedDataValue>>(v: &T) -> Result<Option<OwnedDataValue>> {
475    Ok(Some(v.clone().into()))
476}
477
478fn scalar<T: serde::de::DeserializeOwned>(
479    raw: &Value,
480    cx: &mut CompileCx<'_>,
481    expected: &str,
482) -> Option<T> {
483    match T::deserialize(raw) {
484        Ok(v) => Some(v),
485        Err(_) => {
486            cx.issue(IssueCode::InputParse, format!("expected {expected}"));
487            None
488        }
489    }
490}
491
492macro_rules! scalar_field {
493    ($($t:ty => $kind:ident, $expected:expr;)*) => {$(
494        impl ConfigField for $t {
495            fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
496                scalar(raw, cx, $expected)
497            }
498            fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
499                owned(self)
500            }
501            fn kind() -> FieldKind {
502                FieldKind::Literal(ValueKind::$kind)
503            }
504        }
505    )*};
506}
507
508scalar_field! {
509    bool => Bool, "a boolean";
510    String => String, "a string";
511    u8 => Number, "an unsigned integer that fits u8";
512    u16 => Number, "an unsigned integer that fits u16";
513    u32 => Number, "an unsigned integer that fits u32";
514    u64 => Number, "an unsigned integer that fits u64";
515    usize => Number, "an unsigned integer that fits usize";
516    i32 => Number, "an integer that fits i32";
517    i64 => Number, "an integer that fits i64";
518}
519
520/// A float, written as one: `2.0` stays `2.0`, as the JSON bridge keeps a
521/// float literal, where [`OwnedDataValue`]'s `From<f64>` would make it the
522/// integer `2`.
523impl ConfigField for f64 {
524    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
525        scalar(raw, cx, "a number")
526    }
527    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
528        Ok(Some(OwnedDataValue::Number(NumberValue::Float(*self))))
529    }
530    fn kind() -> FieldKind {
531        FieldKind::Literal(ValueKind::Number)
532    }
533}
534
535impl ConfigField for OwnedDataValue {
536    const NULL_IS_VALUE: bool = true;
537    fn compile(raw: &Value, _cx: &mut CompileCx<'_>) -> Option<Self> {
538        Some(Self::from(raw))
539    }
540    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
541        owned(self)
542    }
543    fn kind() -> FieldKind {
544        FieldKind::Literal(ValueKind::Any)
545    }
546}
547
548impl ConfigField for Value {
549    const NULL_IS_VALUE: bool = true;
550    fn compile(raw: &Value, _cx: &mut CompileCx<'_>) -> Option<Self> {
551        Some(raw.clone())
552    }
553    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
554        Ok(Some(OwnedDataValue::from(self)))
555    }
556    fn kind() -> FieldKind {
557        FieldKind::Literal(ValueKind::Any)
558    }
559}
560
561impl<T: FromValue> ConfigField for Expr<T> {
562    /// `null` is a constant expression.
563    const NULL_IS_VALUE: bool = true;
564    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
565        match cx.exprs().compile::<T>(raw) {
566            Ok(expr) => Some(expr),
567            Err(error) => {
568                // `INVALID_PATH` is a literal string that is not a path; a
569                // non-string where a path goes is a type error.
570                let code = match error.kind() {
571                    ErrorKind::Config if T::KIND == ValueKind::Path && raw.is_string() => {
572                        IssueCode::InvalidPath
573                    }
574                    ErrorKind::Config => IssueCode::InputParse,
575                    _ => IssueCode::TemplateCompile,
576                };
577                cx.issue(code, error.message().to_owned());
578                None
579            }
580        }
581    }
582    fn visit(&self, v: &mut Visit<'_>) {
583        v.expr(self.raw());
584    }
585    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
586        r.eval(self).map(Some)
587    }
588    fn kind() -> FieldKind {
589        FieldKind::Expr(T::KIND)
590    }
591}
592
593impl<T: ConfigField> ConfigField for Option<T> {
594    /// `null` is `None`, or `T`'s own `null`.
595    const NULL_IS_VALUE: bool = true;
596    /// An `Option<Output>` is an output too.
597    const IS_OUTPUT: bool = T::IS_OUTPUT;
598    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
599        if raw.is_null() && !T::NULL_IS_VALUE {
600            return Some(None);
601        }
602        T::compile(raw, cx).map(Some)
603    }
604    fn absent() -> Option<Self> {
605        Some(None)
606    }
607    fn visit(&self, v: &mut Visit<'_>) {
608        if let Some(inner) = self {
609            inner.visit(v);
610        }
611    }
612    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
613        match self {
614            Some(inner) => inner.resolve(r),
615            None => Ok(None),
616        }
617    }
618    fn kind() -> FieldKind {
619        T::kind()
620    }
621}
622
623impl<T: ConfigField> ConfigField for Vec<T> {
624    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
625        let Some(items) = raw.as_array() else {
626            cx.issue(IssueCode::InputParse, "expected an array");
627            return None;
628        };
629        // Every element is compiled, so every issue is reported.
630        let mut out = Self::with_capacity(items.len());
631        let mut ok = true;
632        for (i, item) in items.iter().enumerate() {
633            match cx.at(i, |cx| T::compile(item, cx)) {
634                Some(v) => out.push(v),
635                None => ok = false,
636            }
637        }
638        ok.then_some(out)
639    }
640    fn visit(&self, v: &mut Visit<'_>) {
641        for (i, item) in self.iter().enumerate() {
642            v.at(i, |v| item.visit(v));
643        }
644    }
645    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
646        let items = self
647            .iter()
648            .map(|item| Ok(item.resolve(r)?.unwrap_or(OwnedDataValue::Null)))
649            .collect::<Result<Vec<_>>>()?;
650        Ok(Some(OwnedDataValue::Array(items)))
651    }
652    fn kind() -> FieldKind {
653        FieldKind::List(T::kind)
654    }
655}
656
657impl<T: ConfigField> ConfigField for BTreeMap<String, T> {
658    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
659        let Some(entries) = raw.as_object() else {
660            cx.issue(IssueCode::InputParse, "expected an object");
661            return None;
662        };
663        let mut out = Self::new();
664        let mut ok = true;
665        for (key, value) in entries {
666            match cx.at(key.as_str(), |cx| T::compile(value, cx)) {
667                Some(v) => {
668                    out.insert(key.clone(), v);
669                }
670                None => ok = false,
671            }
672        }
673        ok.then_some(out)
674    }
675    fn visit(&self, v: &mut Visit<'_>) {
676        for (key, item) in self {
677            v.at(key.as_str(), |v| item.visit(v));
678        }
679    }
680    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
681        let entries = self
682            .iter()
683            .map(|(key, item)| {
684                let value = item.resolve(r)?.unwrap_or(OwnedDataValue::Null);
685                Ok((key.clone(), value))
686            })
687            .collect::<Result<Vec<_>>>()?;
688        Ok(Some(OwnedDataValue::Object(entries)))
689    }
690    fn kind() -> FieldKind {
691        FieldKind::Map(T::kind)
692    }
693}
694
695#[cfg(test)]
696mod tests {
697    use super::*;
698
699    #[test]
700    fn adopt_puts_an_issue_below_the_current_path_with_the_step_and_field() {
701        let engine = ExprEngine::default();
702        let mut issues = Vec::new();
703        let rejected = {
704            let mut cx = CompileCx::new(&engine, &mut issues)
705                .at_path("tasks[1].function.input".parse().unwrap())
706                .for_step(StepKind::Task, "fetch");
707            assert_eq!(cx.rejected(), 0);
708            cx.adopt(Issue::new(
709                IssueCode::InputParse,
710                "connector".parse().unwrap(),
711                "no connector",
712            ));
713            // An Advisory is adopted and not counted.
714            cx.adopt(Issue::new(
715                IssueCode::UnknownKey,
716                "retries".parse().unwrap(),
717                "unknown",
718            ));
719            // The path is restored: the adopted issues did not move it.
720            assert_eq!(cx.path().to_string(), "tasks[1].function.input");
721            cx.rejected()
722        };
723        assert_eq!(rejected, 1);
724        assert_eq!(issues.len(), 2);
725        let issue = &issues[0];
726        assert_eq!(issue.code, IssueCode::InputParse);
727        assert_eq!(issue.severity, Severity::Rejected);
728        assert_eq!(issue.path.to_string(), "tasks[1].function.input.connector");
729        assert_eq!(issue.field.as_deref(), Some("connector"));
730        assert_eq!(issue.step_kind, Some(StepKind::Task));
731        assert_eq!(issue.step_id.as_deref(), Some("fetch"));
732        assert_eq!(issue.message, "no connector");
733        assert_eq!(issues[1].severity, Severity::Advisory);
734        assert_eq!(issues[1].field.as_deref(), Some("retries"));
735    }
736}
dataflow_rs/expr/config.rs

dataflow_rs/expr/
config.rs
