dataflow_rs/expr/

config.rs

//! Config fields: the types a function's `input` is made of, and the context
//! that compiles them with every issue at its own path.

use super::document::Interpolated;
use super::engine::ExprEngine;
use super::expr::{Expr, FromValue, ResolveExpr};
use super::output::Output;
use crate::base::typo::unknown_key;
use crate::base::{AuthoredPath, ErrorKind, Issue, IssueCode, PathSeg, Result, Severity, StepKind};
use datavalue::{NumberValue, OwnedDataValue};
use serde_json::Value;
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::fmt;

/// What a field's value is, for a field's schema and for checking a literal.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum ValueKind {
    /// Any JSON.
    Any,
    /// A string.
    String,
    /// A number.
    Number,
    /// A boolean.
    Bool,
    /// An object.
    Object,
    /// An array.
    Array,
    /// A context path, rooted at `data`, `metadata` or `temp_data`.
    Path,
}

/// What a config field is, for tooling.
#[derive(Clone, Copy, Debug)]
#[non_exhaustive]
pub enum FieldKind {
    /// A literal of the given kind, never evaluated.
    Literal(ValueKind),
    /// JSONLogic resolving to the given kind.
    Expr(ValueKind),
    /// A document with expression holes.
    Interpolated,
    /// A write destination.
    Output,
    /// A runtime-declared set of fields.
    Templated,
    /// An array of the element kind.
    List(fn() -> Self),
    /// An object whose values are of the given kind.
    Map(fn() -> Self),
    /// A nested config's fields.
    Nested(fn() -> &'static [FieldSpec]),
}

impl FieldKind {
    /// Whether the kind is an output, or holds one at any depth of a list or
    /// a map. The engine records every output's result whatever its field
    /// says, so a field of such a kind records. A nested config's
    /// own fields say so for themselves.
    pub(crate) fn contains_output(self) -> bool {
        match self {
            Self::Output => true,
            Self::List(element) | Self::Map(element) => element().contains_output(),
            _ => false,
        }
    }
}

/// One field of a config.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct FieldSpec {
    /// The authored key.
    pub name: Cow<'static, str>,
    /// What the field is.
    pub kind: FieldKind,
    /// Is the key required?
    pub required: bool,
    /// Does the engine record the field's result? Always for a field holding
    /// an output.
    pub records: bool,
    /// The default, as JSON text.
    pub default: Option<Cow<'static, str>>,
    /// Is this the primary output?
    pub primary: bool,
    /// The field's doc comment.
    pub description: Option<Cow<'static, str>>,
}

impl FieldSpec {
    /// A runtime field: its kind is an expression, an interpolated document, a
    /// literal or an output. It records when it holds an output.
    pub fn runtime(name: impl Into<Cow<'static, str>>, kind: FieldKind, required: bool) -> Self {
        Self {
            name: name.into(),
            kind,
            required,
            records: kind.contains_output(),
            default: None,
            primary: false,
            description: None,
        }
    }

    /// Sets the default, as JSON text.
    #[must_use]
    pub fn with_default(mut self, json: impl Into<Cow<'static, str>>) -> Self {
        self.default = Some(json.into());
        self
    }
}

/// Receives every expression of a config, with its path.
pub trait ExprVisitor {
    /// An expression at `at`; `records` is set when the engine records its
    /// result.
    fn expr(&mut self, _at: &AuthoredPath, _raw: &Value, _records: bool) {}

    /// A compiled expression at `at`; forwards to [`Self::expr`] by default.
    #[doc(hidden)]
    fn compiled_expr(&mut self, at: &AuthoredPath, expr: &dyn ResolveExpr, records: bool) {
        self.expr(at, expr.raw(), records);
    }

    /// An interpolated document at `at`.
    fn interpolated(&mut self, _at: &AuthoredPath, _doc: &Interpolated) {}
}

/// A walk over a compiled config's expressions.
pub struct Visit<'v> {
    visitor: &'v mut dyn ExprVisitor,
    path: AuthoredPath,
    records: bool,
}

impl<'v> Visit<'v> {
    /// A walk reporting to `visitor`, with paths below `root`.
    pub fn new(visitor: &'v mut dyn ExprVisitor, root: AuthoredPath) -> Self {
        Self {
            visitor,
            path: root,
            records: false,
        }
    }

    /// Runs `f` one segment deeper.
    pub fn at(&mut self, seg: impl Into<PathSeg>, f: impl FnOnce(&mut Self)) {
        self.path.push(seg);
        f(self);
        self.path.pop();
    }

    /// Runs `f` with the `records` flag ORed in, restoring it after.
    pub fn recording(&mut self, records: bool, f: impl FnOnce(&mut Self)) {
        let before = self.records;
        self.records |= records;
        f(self);
        self.records = before;
    }

    /// Reports an expression at the current path.
    pub fn expr(&mut self, raw: &Value) {
        self.visitor.expr(&self.path, raw, self.records);
    }

    /// Reports a compiled expression at the current path.
    pub(crate) fn compiled_expr(&mut self, expr: &dyn ResolveExpr) {
        self.visitor.compiled_expr(&self.path, expr, self.records);
    }

    /// Reports an interpolated document at the current path.
    pub fn interpolated(&mut self, doc: &Interpolated) {
        self.visitor.interpolated(&self.path, doc);
    }
}

impl fmt::Debug for Visit<'_> {
    /// The current path and `records` flag; the visitor is not shown.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Visit")
            .field("path", &format_args!("{}", self.path))
            .field("records", &self.records)
            .finish_non_exhaustive()
    }
}

/// Compiles config fields, recording each issue at its exact authored path.
pub struct CompileCx<'a> {
    exprs: &'a ExprEngine,
    issues: &'a mut Vec<Issue>,
    path: AuthoredPath,
    root_depth: usize,
    step: Option<(StepKind, String)>,
    runtime_fields: Option<&'a [FieldSpec]>,
    rejected: usize,
}

impl<'a> CompileCx<'a> {
    /// A context rooted at the document itself.
    pub fn new(exprs: &'a ExprEngine, issues: &'a mut Vec<Issue>) -> Self {
        Self {
            exprs,
            issues,
            path: AuthoredPath::root(),
            root_depth: 0,
            step: None,
            runtime_fields: None,
            rejected: 0,
        }
    }

    /// Roots the walk at `path`; a field is the first key below it.
    #[must_use]
    pub fn at_path(mut self, path: AuthoredPath) -> Self {
        self.root_depth = path.segs().len();
        self.path = path;
        self
    }

    /// Stamps the step on every issue.
    #[must_use]
    pub fn for_step(mut self, kind: StepKind, id: impl Into<String>) -> Self {
        self.step = Some((kind, id.into()));
        self
    }

    /// The fields a plugin declared at runtime.
    #[must_use]
    pub fn with_runtime_fields(mut self, fields: &'a [FieldSpec]) -> Self {
        self.runtime_fields = Some(fields);
        self
    }

    /// The engine expressions compile with.
    pub fn exprs(&self) -> &ExprEngine {
        self.exprs
    }

    /// The current path.
    pub fn path(&self) -> &AuthoredPath {
        &self.path
    }

    /// The runtime-declared fields, if any.
    pub fn runtime_fields(&self) -> Option<&'a [FieldSpec]> {
        self.runtime_fields
    }

    /// Records an issue at the current path, at the code's default severity.
    pub fn issue(&mut self, code: IssueCode, message: impl Into<String>) {
        self.issue_as(code, code.severity(), message);
    }

    /// Records an issue at the current path with an explicit severity.
    pub fn issue_as(&mut self, code: IssueCode, severity: Severity, message: impl Into<String>) {
        let issue = Issue::new(code, self.path.clone(), message).with_severity(severity);
        self.record(issue);
    }

    /// Records an issue found by another walk, such as a function's own
    /// checks, whose path is relative to the current path. The path is
    /// prefixed with the current one, the step and the field are stamped as
    /// [`Self::issue`] stamps them, and a Rejected issue is counted.
    pub(crate) fn adopt(&mut self, mut issue: Issue) {
        issue.path = self.path.extend(&issue.path);
        self.record(issue);
    }

    /// Stamps the step and the field on `issue`, counts it when it is
    /// Rejected, and records it.
    fn record(&mut self, mut issue: Issue) {
        if let Some((kind, id)) = &self.step {
            issue = issue.with_step(*kind, Some(id));
        }
        if let Some(PathSeg::Key(key)) = issue.path.segs().get(self.root_depth) {
            issue.field = Some(key.to_string());
        }
        if issue.severity == Severity::Rejected {
            self.rejected += 1;
        }
        self.issues.push(issue);
    }

    /// Runs `f` one segment deeper.
    pub fn at<R>(&mut self, seg: impl Into<PathSeg>, f: impl FnOnce(&mut Self) -> R) -> R {
        self.path.push(seg);
        let out = f(self);
        self.path.pop();
        out
    }

    /// How many Rejected issues this context has recorded.
    pub fn rejected(&self) -> usize {
        self.rejected
    }

    /// Records an unknown `key` at the current path (the caller is inside
    /// `at(key, …)`): a Rejected `LIKELY_TYPO` naming the near key, else a
    /// Rejected `UNKNOWN_KEY`.
    #[doc(hidden)]
    pub fn unknown_key(&mut self, key: &str, known: &[&str]) {
        let (code, message) = unknown_key(key, known);
        self.issue_as(code, Severity::Rejected, message);
    }
}

impl fmt::Debug for CompileCx<'_> {
    /// The current path, the step, and how many issues are recorded and
    /// Rejected; not the issues or the engine.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CompileCx")
            .field("path", &format_args!("{}", self.path))
            .field("step", &self.step)
            .field("issues", &self.issues.len())
            .field("rejected", &self.rejected)
            .finish_non_exhaustive()
    }
}

/// What [`Resolver`] is sealed with: only the crate implements it.
pub(crate) mod sealed {
    /// The seal: a type the crate resolves configs with.
    pub trait Sealed {}
}

/// What a [`Resolver`] supplies for a config's primary output in place of
/// evaluating it.
///
/// A call's context answers this only when its input is resolved with the
/// destination already known
/// ([`Ctx::input_with_primary_output`](crate::Ctx::input_with_primary_output)
/// and [`Ctx::input_without_primary_output`](crate::Ctx::input_without_primary_output));
/// every other resolution, [`Ctx::input`](crate::Ctx::input) included, leaves
/// every output to resolve as usual.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub enum OutputOverride {
    /// The field is left out, as an optional output the input left unset is.
    Omit,
    /// The field resolves to this value, its expression never evaluated.
    Value(OwnedDataValue),
}

/// What resolving a config needs from a call: an expression evaluated, and
/// a document rendered, each to an owned value, against the context the call
/// sees. A [`ConfigField`] calls it from [`ConfigField::resolve`]; a call's
/// context implements it ([`Ctx`](crate::Ctx), through a shared reference).
///
/// Sealed: a host calls it, from its own [`ConfigField::resolve`], and never
/// implements it, so a method can join it without a breaking change.
///
/// ```compile_fail
/// use dataflow_rs::Result;
/// use dataflow_rs::datavalue::OwnedDataValue;
/// use dataflow_rs::expr::{Interpolated, ResolveExpr, Resolver};
///
/// struct Mine;
///
/// impl Resolver for Mine {
///     fn eval(&mut self, _expr: &dyn ResolveExpr) -> Result<OwnedDataValue> {
///         Ok(OwnedDataValue::Null)
///     }
///     fn render(&mut self, _doc: &Interpolated) -> Result<OwnedDataValue> {
///         Ok(OwnedDataValue::Null)
///     }
/// }
/// ```
pub trait Resolver: sealed::Sealed {
    /// `expr` evaluated, as an owned value: a number for a `u64`, `i64` or
    /// `f64` expression, a boolean for a `bool` one, a string for a `String`
    /// one, a path's dotted text for a `Path` one, and the value itself for
    /// any other. A folded constant is its value, evaluated never.
    ///
    /// # Errors
    ///
    /// The evaluation's: [`ErrorKind::Budget`] when the operation ceiling is
    /// exceeded, [`ErrorKind::Eval`] for any other failure, a result that is
    /// not of the expression's type included.
    fn eval(&mut self, expr: &dyn ResolveExpr) -> Result<OwnedDataValue>;

    /// `doc` rendered, as an owned tree: each hole's value, or its default.
    ///
    /// # Errors
    ///
    /// None from a call's context, which is every resolver there is today;
    /// the `Result` leaves room for one that may refuse.
    fn render(&mut self, doc: &Interpolated) -> Result<OwnedDataValue>;

    /// The value `out` resolves to in place of being evaluated, when this
    /// resolution supplies the callee's primary output: `None` resolves it
    /// as usual, as every resolver does but the two input methods that
    /// supply it.
    ///
    /// Only the [`Output`] a config's
    /// [`primary_output`](crate::function::FunctionConfig::primary_output)
    /// lends is answered; every other output, a nested one included, is
    /// resolved as usual.
    fn output_override(&mut self, _out: &Output) -> Option<OutputOverride> {
        None
    }
}

/// A type a config is made of.
///
/// An `Option<T>` field is optional: an absent key is `None`, and so is an
/// explicit `null`, unless `T` takes `null` as a value
/// ([`NULL_IS_VALUE`](Self::NULL_IS_VALUE)), as an [`Expr`], an
/// [`Interpolated`] document and any JSON value do. There the `null` is
/// compiled as `T`: `map`'s `"logic": null` is a constant `null` expression,
/// not a missing `logic`.
///
/// A host writes its own field type by implementing it: here a connector's
/// name, a newtype over an `Expr<String>`, compiled, walked and resolved as
/// the expression is, so a derived config holding one checks its secrets
/// and template keys, and resolves it, as it would the expression.
///
/// ```
/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
/// use dataflow_rs::datavalue::OwnedDataValue;
/// use dataflow_rs::expr::{CompileCx, ConfigField, FieldKind, Resolver, ValueKind, Visit};
/// use dataflow_rs::function::{FunctionConfig, testing};
/// use dataflow_rs::{Ctx, Expr, Function, Message, Outcome, Result};
/// use serde_json::{Value, json};
///
/// /// The connector a call reaches, by name.
/// struct Connector(Expr<String>);
///
/// impl ConfigField for Connector {
///     fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
///         Expr::<String>::compile(raw, cx).map(Self)
///     }
///     fn visit(&self, v: &mut Visit<'_>) {
///         v.expr(self.0.raw());
///     }
///     fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
///         r.eval(&self.0).map(Some)
///     }
///     fn kind() -> FieldKind {
///         FieldKind::Expr(ValueKind::String)
///     }
/// }
///
/// #[derive(FunctionConfig)]
/// struct Query {
///     connector: Connector,
///     limit: Option<Expr<u64>>,
/// }
///
/// /// Writes its input, as a layer or a dry run reads it, at `data.input`.
/// struct Echo;
///
/// impl Function for Echo {
///     type Config = Query;
///
///     async fn call(&self, ctx: &mut Ctx<'_>, _cfg: &Query) -> Result<Outcome> {
///         let input = ctx.input()?;
///         ctx.set(&"data.input".parse()?, input)?;
///         Ok(Outcome::Ok)
///     }
/// }
///
/// let input = json!({"connector": {"cat": ["db-", {"var": "data.region"}]}});
/// let data = OwnedDataValue::from(&json!({"region": "eu"}));
/// let out = testing::call(Echo, input)
///     .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
///     .await;
/// assert_eq!(out.outcome.unwrap(), Outcome::Ok);
/// // The field resolved at its key; the `Option` left unset is absent.
/// let at = "data.input".parse().unwrap();
/// let written = out.message.context().get(&at).unwrap();
/// assert_eq!(serde_json::to_string(written).unwrap(), r#"{"connector":"db-eu"}"#);
/// # });
/// ```
pub trait ConfigField: Sized + Send + Sync + 'static {
    /// Whether `null` is one of the type's values. When it is not, an explicit
    /// `null` in an `Option<Self>` field means `None`.
    const NULL_IS_VALUE: bool = false;

    /// Whether the field is an output: a destination its call writes to.
    /// `false` unless the type is [`Output`](crate::expr::Output) or an
    /// `Option` of one. The derive reads a field typed `Output` or marked
    /// `#[output]` as an output and asserts that this is `true`, so a type
    /// that merely shares the name is refused at compile time.
    const IS_OUTPUT: bool = false;

    /// Compiles `raw`; `None` after recording at least one issue.
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self>;

    /// The value when the key is absent and no default is declared. `None`
    /// makes the field required.
    fn absent() -> Option<Self> {
        None
    }

    /// Reports the field's expressions.
    fn visit(&self, _v: &mut Visit<'_>) {}

    /// The field's value for this call, resolved through `r`, or `None` when
    /// it has none (an `Option` the input left unset, with no default). The
    /// value sits where [`Self::visit`] reports the field's expressions: a
    /// list's elements at their indices, a map's values at their keys, and
    /// an element left unset is `null`, keeping its place.
    ///
    /// # Errors
    ///
    /// The resolver's, for the first expression or document that fails.
    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>>;

    /// What the field is.
    fn kind() -> FieldKind;
}

/// A scalar's value: itself.
fn owned<T: Clone + Into<OwnedDataValue>>(v: &T) -> Result<Option<OwnedDataValue>> {
    Ok(Some(v.clone().into()))
}

fn scalar<T: serde::de::DeserializeOwned>(
    raw: &Value,
    cx: &mut CompileCx<'_>,
    expected: &str,
) -> Option<T> {
    match T::deserialize(raw) {
        Ok(v) => Some(v),
        Err(_) => {
            cx.issue(IssueCode::InputParse, format!("expected {expected}"));
            None
        }
    }
}

macro_rules! scalar_field {
    ($($t:ty => $kind:ident, $expected:expr;)*) => {$(
        impl ConfigField for $t {
            fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
                scalar(raw, cx, $expected)
            }
            fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
                owned(self)
            }
            fn kind() -> FieldKind {
                FieldKind::Literal(ValueKind::$kind)
            }
        }
    )*};
}

scalar_field! {
    bool => Bool, "a boolean";
    String => String, "a string";
    u8 => Number, "an unsigned integer that fits u8";
    u16 => Number, "an unsigned integer that fits u16";
    u32 => Number, "an unsigned integer that fits u32";
    u64 => Number, "an unsigned integer that fits u64";
    usize => Number, "an unsigned integer that fits usize";
    i32 => Number, "an integer that fits i32";
    i64 => Number, "an integer that fits i64";
}

/// A float, written as one: `2.0` stays `2.0`, as the JSON bridge keeps a
/// float literal, where [`OwnedDataValue`]'s `From<f64>` would make it the
/// integer `2`.
impl ConfigField for f64 {
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
        scalar(raw, cx, "a number")
    }
    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        Ok(Some(OwnedDataValue::Number(NumberValue::Float(*self))))
    }
    fn kind() -> FieldKind {
        FieldKind::Literal(ValueKind::Number)
    }
}

impl ConfigField for OwnedDataValue {
    const NULL_IS_VALUE: bool = true;
    fn compile(raw: &Value, _cx: &mut CompileCx<'_>) -> Option<Self> {
        Some(Self::from(raw))
    }
    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        owned(self)
    }
    fn kind() -> FieldKind {
        FieldKind::Literal(ValueKind::Any)
    }
}

impl ConfigField for Value {
    const NULL_IS_VALUE: bool = true;
    fn compile(raw: &Value, _cx: &mut CompileCx<'_>) -> Option<Self> {
        Some(raw.clone())
    }
    fn resolve(&self, _r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        Ok(Some(OwnedDataValue::from(self)))
    }
    fn kind() -> FieldKind {
        FieldKind::Literal(ValueKind::Any)
    }
}

impl<T: FromValue> ConfigField for Expr<T> {
    /// `null` is a constant expression.
    const NULL_IS_VALUE: bool = true;
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
        match cx.exprs().compile::<T>(raw) {
            Ok(expr) => Some(expr),
            Err(error) => {
                // `INVALID_PATH` is a literal string that is not a path; a
                // non-string where a path goes is a type error.
                let code = match error.kind() {
                    ErrorKind::Config if T::KIND == ValueKind::Path && raw.is_string() => {
                        IssueCode::InvalidPath
                    }
                    ErrorKind::Config => IssueCode::InputParse,
                    _ => IssueCode::TemplateCompile,
                };
                cx.issue(code, error.message().to_owned());
                None
            }
        }
    }
    fn visit(&self, v: &mut Visit<'_>) {
        v.compiled_expr(self);
    }
    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        r.eval(self).map(Some)
    }
    fn kind() -> FieldKind {
        FieldKind::Expr(T::KIND)
    }
}

impl<T: ConfigField> ConfigField for Option<T> {
    /// `null` is `None`, or `T`'s own `null`.
    const NULL_IS_VALUE: bool = true;
    /// An `Option<Output>` is an output too.
    const IS_OUTPUT: bool = T::IS_OUTPUT;
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
        if raw.is_null() && !T::NULL_IS_VALUE {
            return Some(None);
        }
        T::compile(raw, cx).map(Some)
    }
    fn absent() -> Option<Self> {
        Some(None)
    }
    fn visit(&self, v: &mut Visit<'_>) {
        if let Some(inner) = self {
            inner.visit(v);
        }
    }
    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        match self {
            Some(inner) => inner.resolve(r),
            None => Ok(None),
        }
    }
    fn kind() -> FieldKind {
        T::kind()
    }
}

impl<T: ConfigField> ConfigField for Vec<T> {
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
        let Some(items) = raw.as_array() else {
            cx.issue(IssueCode::InputParse, "expected an array");
            return None;
        };
        // Every element is compiled, so every issue is reported.
        let mut out = Self::with_capacity(items.len());
        let mut ok = true;
        for (i, item) in items.iter().enumerate() {
            match cx.at(i, |cx| T::compile(item, cx)) {
                Some(v) => out.push(v),
                None => ok = false,
            }
        }
        ok.then_some(out)
    }
    fn visit(&self, v: &mut Visit<'_>) {
        for (i, item) in self.iter().enumerate() {
            v.at(i, |v| item.visit(v));
        }
    }
    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        let items = self
            .iter()
            .map(|item| Ok(item.resolve(r)?.unwrap_or(OwnedDataValue::Null)))
            .collect::<Result<Vec<_>>>()?;
        Ok(Some(OwnedDataValue::Array(items)))
    }
    fn kind() -> FieldKind {
        FieldKind::List(T::kind)
    }
}

impl<T: ConfigField> ConfigField for BTreeMap<String, T> {
    fn compile(raw: &Value, cx: &mut CompileCx<'_>) -> Option<Self> {
        let Some(entries) = raw.as_object() else {
            cx.issue(IssueCode::InputParse, "expected an object");
            return None;
        };
        let mut out = Self::new();
        let mut ok = true;
        for (key, value) in entries {
            match cx.at(key.as_str(), |cx| T::compile(value, cx)) {
                Some(v) => {
                    out.insert(key.clone(), v);
                }
                None => ok = false,
            }
        }
        ok.then_some(out)
    }
    fn visit(&self, v: &mut Visit<'_>) {
        for (key, item) in self {
            v.at(key.as_str(), |v| item.visit(v));
        }
    }
    fn resolve(&self, r: &mut dyn Resolver) -> Result<Option<OwnedDataValue>> {
        let entries = self
            .iter()
            .map(|(key, item)| {
                let value = item.resolve(r)?.unwrap_or(OwnedDataValue::Null);
                Ok((key.clone(), value))
            })
            .collect::<Result<Vec<_>>>()?;
        Ok(Some(OwnedDataValue::Object(entries)))
    }
    fn kind() -> FieldKind {
        FieldKind::Map(T::kind)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn adopt_puts_an_issue_below_the_current_path_with_the_step_and_field() {
        let engine = ExprEngine::default();
        let mut issues = Vec::new();
        let rejected = {
            let mut cx = CompileCx::new(&engine, &mut issues)
                .at_path("tasks[1].function.input".parse().unwrap())
                .for_step(StepKind::Task, "fetch");
            assert_eq!(cx.rejected(), 0);
            cx.adopt(Issue::new(
                IssueCode::InputParse,
                "connector".parse().unwrap(),
                "no connector",
            ));
            // An Advisory is adopted and not counted.
            cx.adopt(Issue::new(
                IssueCode::UnknownKey,
                "retries".parse().unwrap(),
                "unknown",
            ));
            // The path is restored: the adopted issues did not move it.
            assert_eq!(cx.path().to_string(), "tasks[1].function.input");
            cx.rejected()
        };
        assert_eq!(rejected, 1);
        assert_eq!(issues.len(), 2);
        let issue = &issues[0];
        assert_eq!(issue.code, IssueCode::InputParse);
        assert_eq!(issue.severity, Severity::Rejected);
        assert_eq!(issue.path.to_string(), "tasks[1].function.input.connector");
        assert_eq!(issue.field.as_deref(), Some("connector"));
        assert_eq!(issue.step_kind, Some(StepKind::Task));
        assert_eq!(issue.step_id.as_deref(), Some("fetch"));
        assert_eq!(issue.message, "no connector");
        assert_eq!(issues[1].severity, Severity::Advisory);
        assert_eq!(issues[1].field.as_deref(), Some("retries"));
    }
}