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

dataflow_rs/expr/
config.rs
