mpl_lang/
query.rs

1//! The query structures
2use std::{
3    collections::{HashMap, HashSet},
4    fmt::Display,
5    num::TryFromIntError,
6};
7
8#[cfg(feature = "clock")]
9use chrono::Utc;
10use chrono::{DateTime, Duration, FixedOffset};
11use miette::SourceSpan;
12use pest::Parser as _;
13use strumbra::SharedString;
14
15use crate::{
16    ParseError,
17    enc_regex::EncodableRegex,
18    linker::{AlignFunction, ComputeFunction, GroupFunction, MapFunction},
19    parser::{self, MPLParser, ParseParamError, Rule},
20    tags::TagValue,
21    time::{Resolution, ResolutionError},
22    types::{BucketSpec, BucketType, Dataset, Metric, Parameterized},
23};
24
25mod fmt;
26#[cfg(test)]
27mod tests;
28
29/// Metric identifier
30#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
31pub struct MetricId {
32    /// The dataset identifier or param
33    pub dataset: Parameterized<Dataset>,
34    /// The metric identifier
35    pub metric: Metric,
36}
37
38/// Time unit
39#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
40pub enum TimeUnit {
41    /// Millisecond
42    Millisecond,
43    /// Second
44    Second,
45    /// Minute
46    Minute,
47    /// Hour
48    Hour,
49    /// Day
50    Day,
51    /// Week
52    Week,
53    /// Month
54    Month,
55    /// Year
56    Year,
57}
58
59#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
60/// Relative time (1h)
61pub struct RelativeTime {
62    /// Value
63    pub value: u64,
64    /// Unit
65    pub unit: TimeUnit,
66}
67
68/// A point in time
69#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
70pub enum Time {
71    /// A time relative to now
72    Relative(RelativeTime),
73    /// A timestamp
74    Timestamp(i64),
75    /// A RFC3339 timestamp
76    RFC3339(DateTime<FixedOffset>),
77    /// A time modifier
78    Modifier(String),
79}
80
81/// A timerange between two times
82#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
83pub struct TimeRange {
84    /// Start time of the range
85    pub start: Time,
86    /// End time of the range or None for 'now'
87    pub end: Option<Time>,
88}
89
90/// The source for a query
91#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
92pub struct Source {
93    /// The metric
94    pub metric_id: MetricId,
95    /// The time range
96    pub time: Option<TimeRange>,
97}
98impl Source {
99    fn time(&self) -> Option<&TimeRange> {
100        self.time.as_ref()
101    }
102}
103
104/// An error related to value parsing
105#[derive(Debug, thiserror::Error)]
106pub enum ValueError {
107    /// Invalid float value
108    #[error("Invalid Float")]
109    BadFloat,
110}
111/// An fragment of a string expression
112#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
113pub enum StringFragment {
114    /// Plain text
115    Text(String),
116    /// Interpolated expression
117    Expr(Expr),
118}
119/// An expression
120#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
121pub enum Expr {
122    /// Constant value Leave
123    Const(TagValue),
124    /// Parameter value
125    Param {
126        /// The location where the param is used
127        span: SourceSpan,
128        /// The param
129        param: ParamDeclaration,
130    },
131    /// A possibly interpolated string value
132    String(Vec<StringFragment>),
133    /// An array
134    Array(Vec<Expr>),
135    /// A reference to a tag value
136    Tag(String),
137}
138
139/// A comparison operator for filtering based on a value
140#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
141pub enum Cmp {
142    /// Equal to the given value
143    Eq(Expr),
144    /// Not equal to the given value
145    Ne(Expr),
146    /// Greater than the given value
147    Gt(Expr),
148    /// Greater than or equal to the given value
149    Ge(Expr),
150    /// Less than the given value
151    Lt(Expr),
152    /// Less than or equal to the given value
153    Le(Expr),
154    /// Is the given tag value in the given list
155    In(Expr),
156    /// Matches the given regular expression
157    RegEx(Parameterized<EncodableRegex>),
158    /// Does not match the given regular expression
159    RegExNot(Parameterized<EncodableRegex>),
160    /// Is the given tag type
161    Is(TagType),
162}
163
164/// Rename the output as a new metric
165#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
166pub struct As {
167    /// The new name for the metric
168    pub name: Metric,
169}
170
171/// Filter the series
172#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
173pub enum Filter {
174    /// Logical AND of the given filters
175    And(Vec<Filter>),
176    /// Logical OR of the given filters
177    Or(Vec<Filter>),
178    /// Logical NOT of the given filters
179    Not(Box<Filter>),
180    /// Filter based on a field
181    Cmp {
182        /// The field to filter on
183        field: String,
184        /// The comparison to perform
185        rhs: Cmp,
186    },
187}
188
189/// Ifdef conditionally filters the series
190#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
191pub enum FilterOrIfDef {
192    /// A plain filter
193    Filter(Filter),
194    /// ifdef based on a parameter declaration
195    Ifdef {
196        /// The name of the parameter
197        param: ParamDeclaration,
198        /// The filter
199        filter: Filter,
200        /// The else filter
201        else_filter: Option<Filter>,
202    },
203}
204
205impl FilterOrIfDef {
206    #[cfg(test)]
207    pub(crate) fn filter(&self) -> &Filter {
208        match self {
209            FilterOrIfDef::Filter(filter) | FilterOrIfDef::Ifdef { filter, .. } => filter,
210        }
211    }
212}
213
214/// A Mapping function
215#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
216pub struct Mapping {
217    /// The function to apply
218    pub function: MapFunction,
219    /// The optional argument to pass to the function
220    pub arg: Option<f64>,
221}
222
223/// An Alignment function
224#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
225pub struct Align {
226    /// The function to apply
227    pub function: AlignFunction,
228    /// The time to align to
229    pub time: Option<Parameterized<RelativeTime>>,
230}
231
232/// A Grouping function
233#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
234pub struct GroupBy {
235    /// The location of the group by clause
236    pub span: SourceSpan,
237    /// The function to apply
238    pub function: GroupFunction,
239    /// The tags to group by
240    pub tags: Vec<String>,
241}
242
243/// A Bucketing function, applying both tag and time based aggregation
244#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
245pub struct BucketBy {
246    /// The location of the group by clause
247    pub span: SourceSpan,
248    /// The function to apply
249    pub function: BucketType,
250    /// The time to align to
251    pub time: Option<Parameterized<RelativeTime>>,
252    /// The tags to group by
253    pub tags: Vec<String>,
254    /// The buckets to produce
255    pub spec: Vec<BucketSpec>,
256}
257
258/// Possible aggregate functions
259#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
260pub enum Aggregate {
261    /// Map a function over each value
262    Map(Mapping),
263    /// Align the data to a time interval
264    Align(Align),
265    /// Group the data by tags
266    GroupBy(GroupBy),
267    /// Bucket the data by time and tags
268    Bucket(BucketBy),
269    /// Rename the metric
270    As(As),
271}
272
273/// Extends a series with a new tag
274#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)]
275pub struct TagExtend {
276    /// The name of the new tag to add
277    pub tag: String,
278    /// The value of the new tag
279    pub value: Expr,
280}
281
282/// Values for directives
283#[cfg_attr(feature = "bincode", derive(bincode::Encode, bincode::Decode))]
284#[derive(Clone, Debug, PartialEq, serde::Deserialize, serde::Serialize)]
285pub enum DirectiveValue {
286    /// Directive with a ident value
287    Ident(String),
288    /// Directive with a literal value
289    Int(i64),
290    /// Directive with a float value
291    Float(f64),
292    /// Directive with a string value
293    String(String),
294    /// Directive with a boolean value
295    Bool(bool),
296    /// Directive with no value
297    None,
298}
299
300impl DirectiveValue {
301    /// Ident value
302    #[must_use]
303    pub fn as_ident(&self) -> Option<&str> {
304        match self {
305            DirectiveValue::Ident(ident) => Some(ident),
306            _ => None,
307        }
308    }
309    /// Int value
310    #[must_use]
311    pub fn as_int(&self) -> Option<i64> {
312        match self {
313            DirectiveValue::Int(int) => Some(*int),
314            _ => None,
315        }
316    }
317    /// Float value
318    #[must_use]
319    pub fn as_float(&self) -> Option<f64> {
320        match self {
321            DirectiveValue::Float(float) => Some(*float),
322            _ => None,
323        }
324    }
325    /// String value
326    #[must_use]
327    pub fn as_string(&self) -> Option<&str> {
328        match self {
329            DirectiveValue::String(string) => Some(string),
330            _ => None,
331        }
332    }
333    /// Bool value
334    #[must_use]
335    pub fn as_bool(&self) -> Option<bool> {
336        match self {
337            DirectiveValue::Bool(bool) => Some(*bool),
338            _ => None,
339        }
340    }
341    /// Tests if value is None
342    #[must_use]
343    pub fn is_none(&self) -> bool {
344        matches!(self, DirectiveValue::None)
345    }
346    /// Tests if value is Some
347    #[must_use]
348    pub fn is_some(&self) -> bool {
349        !self.is_none()
350    }
351}
352
353/// A parameter type, either Optional or Terminal.
354#[derive(Clone, Copy, Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
355pub enum ParamType {
356    /// A type that's defined and present `param p: int`
357    Terminal(TerminalParamType),
358    /// A type that may or may not be present `param p: Option<int>`
359    Optional(TerminalParamType),
360}
361
362impl ParamType {
363    fn is_optional(self) -> bool {
364        matches!(self, ParamType::Optional(_))
365    }
366    fn typ(self) -> TerminalParamType {
367        match self {
368            ParamType::Terminal(terminal_param_type) | ParamType::Optional(terminal_param_type) => {
369                terminal_param_type
370            }
371        }
372    }
373}
374
375impl std::fmt::Display for ParamType {
376    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
377        match self {
378            ParamType::Terminal(t) => t.fmt(f),
379            ParamType::Optional(t) => write!(f, "Option<{t}>"),
380        }
381    }
382}
383
384/// Terminal Types for params.
385#[derive(Clone, Copy, Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
386pub enum TerminalParamType {
387    /// Duration (e.g. 25s)
388    Duration,
389    /// Dataset
390    Dataset,
391    /// Regex
392    Regex,
393    /// A tag value type
394    Tag(TagType),
395}
396impl std::fmt::Display for TerminalParamType {
397    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
398        match self {
399            TerminalParamType::Dataset => write!(f, "Dataset"),
400            TerminalParamType::Duration => write!(f, "Duration"),
401            TerminalParamType::Regex => write!(f, "Regex"),
402            TerminalParamType::Tag(t) => t.fmt(f),
403        }
404    }
405}
406
407/// Types for params.
408#[cfg_attr(feature = "bincode", derive(bincode::Encode, bincode::Decode))]
409#[derive(Clone, Copy, Debug, Hash, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
410pub enum TagType {
411    /// String
412    String,
413    /// Int
414    Int,
415    /// Float
416    Float,
417    /// Bool
418    Bool,
419    /// Null value
420    Null,
421    /// An array of values
422    Array,
423}
424
425#[cfg(feature = "bincode")]
426#[test]
427fn test_renaming_none_to_null_has_no_bincode_side_effects() {
428    let enc = [4];
429    assert_eq!(
430        (TagType::Null, 1),
431        bincode::decode_from_slice(&enc, bincode::config::standard()).expect("it does ...")
432    );
433}
434
435impl std::fmt::Display for TagType {
436    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
437        match self {
438            TagType::String => write!(f, "string"),
439            TagType::Int => write!(f, "int"),
440            TagType::Float => write!(f, "float"),
441            TagType::Bool => write!(f, "bool"),
442            TagType::Null => write!(f, "null"),
443            TagType::Array => write!(f, "array"),
444        }
445    }
446}
447
448/// Directives given to adjust the behavior of the runtime
449pub type Directives = HashMap<String, DirectiveValue>;
450
451/// A param.
452#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
453pub struct ParamDeclaration {
454    /// The location of the param
455    pub span: SourceSpan,
456    /// The name of the param
457    pub name: String,
458    /// The type of the param
459    pub typ: ParamType,
460}
461
462impl ParamDeclaration {
463    pub(crate) fn typ(&self) -> TerminalParamType {
464        self.typ.typ()
465    }
466
467    pub(crate) fn is_optional(&self) -> bool {
468        self.typ.is_optional()
469    }
470}
471
472/// A param value.
473#[derive(Debug, Clone, PartialEq)]
474pub enum ParamValue {
475    /// Dataset
476    Dataset(Dataset),
477    /// Duration
478    Duration(RelativeTime),
479    /// String
480    String(String),
481    /// Int
482    Int(i64),
483    /// Float
484    Float(f64),
485    /// Bool
486    Bool(bool),
487    /// Regex
488    Regex(EncodableRegex),
489    /// Array
490    Array(Vec<TagValue>),
491}
492
493impl ParamValue {
494    /// Get the type of the param value.
495    #[must_use]
496    pub fn typ(&self) -> TerminalParamType {
497        match self {
498            ParamValue::Dataset(_) => TerminalParamType::Dataset,
499            ParamValue::Duration(_) => TerminalParamType::Duration,
500            ParamValue::Regex(_) => TerminalParamType::Regex,
501            ParamValue::String(_) => TerminalParamType::Tag(TagType::String),
502            ParamValue::Int(_) => TerminalParamType::Tag(TagType::Int),
503            ParamValue::Float(_) => TerminalParamType::Tag(TagType::Float),
504            ParamValue::Bool(_) => TerminalParamType::Tag(TagType::Bool),
505            ParamValue::Array(_) => TerminalParamType::Tag(TagType::Array),
506        }
507    }
508}
509
510/// The param provided to the query.
511#[derive(Debug, Clone, PartialEq)]
512pub struct ProvidedParam {
513    /// The name of the param.
514    pub name: String,
515    /// The value.
516    pub value: ParamValue,
517}
518
519impl ProvidedParam {
520    /// Create a new `ProvidedParam`.
521    pub fn new(name: impl Into<String>, value: ParamValue) -> Self {
522        Self {
523            name: name.into(),
524            value,
525        }
526    }
527}
528
529/// A smol wrapper around `Vec<ProvidedParam>` for easier use.
530#[derive(Debug, Clone, Default)]
531pub struct ProvidedParams {
532    inner: Vec<ProvidedParam>,
533}
534
535/// The error returned from `ProvidedParams::resolve`.
536#[derive(Debug, thiserror::Error)]
537pub enum ResolveError {
538    /// Param not provided
539    #[error("Param ${0} was not provided to the query")]
540    ParamNotProvided(String),
541    /// Invalid type
542    #[error(
543        "Param ${name} is defined as `{defined}`, but was used in a context that expected one of: {}",
544        expected.iter().map(ToString::to_string).collect::<Vec<_>>().join(", ")
545    )]
546    InvalidType {
547        /// Name of the param
548        name: String,
549        /// Type of the param
550        defined: TerminalParamType,
551        /// The type that is valid in the context it was used
552        expected: Vec<TerminalParamType>,
553    },
554    /// Shared string error
555    #[error("Shared string error: {0}")]
556    SharedString(#[from] strumbra::Error),
557}
558
559/// The error returned from `ProvidedParams::parse`.
560#[derive(Debug, thiserror::Error)]
561pub enum ParseProvidedParamsError {
562    /// Parse failed
563    #[error("Failed to parse the value for ${param_name} as {expected_type}: {err}")]
564    ParseParam {
565        /// Param name
566        param_name: String,
567        /// Expected t ype
568        expected_type: ParamType,
569        /// Parse param error
570        err: ParseParamError,
571    },
572    /// Params provided more than once
573    #[error("These params were provided more than once: {}", .0.join(", "))]
574    ParamsProvidedMoreThanOnce(Vec<String>),
575    /// Params declared but not provided
576    #[error("The following params were declared but not provided: {}", .0.join(", "))]
577    ParamsDeclaredButNotProvided(Vec<String>),
578    /// Too many params provided
579    #[error("The number of params provided exceeds the upper limit of {0}")]
580    TooManyParamsProvided(usize),
581}
582/// List of warning reasons
583#[derive(Debug)]
584pub enum WarningReason {
585    /// Provided but not declared  param
586    ParamNotDeclared(Vec<String>),
587    /// System parameter declared
588    ParamUsingSystemPrefix {
589        /// The param
590        param: String,
591    },
592    /// lowercase duration
593    OldDuration,
594}
595
596impl Display for WarningReason {
597    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
598        match self {
599            WarningReason::ParamNotDeclared(items) => write!(
600                f,
601                "These params were provided but not declared: {}",
602                items.join(", ")
603            ),
604            WarningReason::OldDuration => {
605                write!(f, "`duration` is depricated, please ues `Duration`")
606            }
607            WarningReason::ParamUsingSystemPrefix { param } => {
608                write!(
609                    f,
610                    "The param ${param} uses the `__` prefix reserved for system params"
611                )
612            }
613        }
614    }
615}
616
617/// Warning we want to surface to the user instead of failing the request.
618#[derive(Debug)]
619pub struct Warning {
620    source: Option<SourceSpan>,
621    warning: WarningReason,
622}
623
624impl Warning {
625    /// The warning message
626    #[must_use]
627    pub fn warning(&self) -> &WarningReason {
628        &self.warning
629    }
630    /// The location of the warning (if any)
631    #[must_use]
632    pub fn source(&self) -> Option<SourceSpan> {
633        self.source
634    }
635}
636
637/// Warnings we want to surface to the user instead of failing the request.
638#[derive(Debug, Default)]
639pub struct Warnings {
640    inner: Vec<Warning>,
641}
642
643impl Warnings {
644    /// Create a new warnings structure.
645    #[must_use]
646    pub fn new() -> Self {
647        Self::default()
648    }
649
650    /// Add a new warning.
651    pub fn push(&mut self, warning: WarningReason) {
652        self.inner.push(Warning {
653            source: None,
654            warning,
655        });
656    }
657    /// Add a new warning.
658    pub fn push_span(&mut self, span: SourceSpan, warning: WarningReason) {
659        self.inner.push(Warning {
660            source: Some(span),
661            warning,
662        });
663    }
664
665    /// Returns true if there are no warnings.
666    #[must_use]
667    pub fn is_empty(&self) -> bool {
668        self.inner.is_empty()
669    }
670
671    /// Get the warnings as slice.
672    #[must_use]
673    pub fn as_slice(&self) -> &[Warning] {
674        &self.inner
675    }
676
677    /// Turn into a vector.
678    #[must_use]
679    pub fn into_vec(self) -> Vec<Warning> {
680        self.inner
681    }
682}
683
684impl ProvidedParams {
685    /// Create a new `ProvidedParams` struct.
686    #[must_use]
687    pub fn new(inner: Vec<ProvidedParam>) -> Self {
688        Self { inner }
689    }
690
691    /// Parse params from a hashmap of query parameters.
692    /// This will only look at params that start with `param__` and it'll use
693    /// the parser definitions to extract the values.
694    pub fn parse_and_validate(
695        mpl_params: &Params,
696        query_params: &[(String, String)],
697    ) -> Result<(Self, Warnings), ParseProvidedParamsError> {
698        const PREFIX: &str = "param__";
699        const PARAM_COUNT_LIMIT: usize = 128;
700
701        let mut warnings = Warnings::new();
702        let mut defined_more_than_once = HashSet::new();
703        let mut provided_but_not_declared = HashSet::new();
704        let mut seen = HashSet::new();
705
706        let params = query_params
707            .iter()
708            .filter_map(|(name, value)| {
709                if !name.starts_with(PREFIX) {
710                    return None;
711                }
712                let name = name.trim_start_matches(PREFIX);
713                if name.is_empty() {
714                    return None;
715                }
716
717                Some((name, value))
718            })
719            .take(PARAM_COUNT_LIMIT + 1)
720            .collect::<Vec<(&str, &String)>>();
721
722        // we don't support unlimited params
723        if params.len() > PARAM_COUNT_LIMIT {
724            return Err(ParseProvidedParamsError::TooManyParamsProvided(
725                PARAM_COUNT_LIMIT,
726            ));
727        }
728
729        let mut provided_params = Vec::new();
730        for (name, value) in params {
731            if seen.contains(name) {
732                // uh oh, we've already seen this value
733                defined_more_than_once.insert(name);
734                continue;
735            }
736            seen.insert(name);
737
738            // is the param even declared?
739            let Some(mpl_param) = mpl_params.iter().find(|p| p.name == name) else {
740                provided_but_not_declared.insert(name);
741                continue;
742            };
743
744            // parse mpl
745            let parsed = MPLParser::parse(Rule::param_value, value).map_err(|err| {
746                ParseProvidedParamsError::ParseParam {
747                    param_name: name.to_string(),
748                    expected_type: mpl_param.typ,
749                    err: ParseParamError::Parse(ParseError::from(err)),
750                }
751            })?;
752
753            // parse as correct type
754            let value = parser::parse_param_value(mpl_param, parsed).map_err(|err| {
755                ParseProvidedParamsError::ParseParam {
756                    param_name: name.to_string(),
757                    expected_type: mpl_param.typ,
758                    err,
759                }
760            })?;
761
762            provided_params.push(ProvidedParam {
763                name: name.to_string(),
764                value,
765            });
766        }
767
768        if !provided_but_not_declared.is_empty() {
769            // sort for consistency
770            let mut items = provided_but_not_declared
771                .into_iter()
772                .map(|p| format!("${p}"))
773                .collect::<Vec<String>>();
774            items.sort();
775
776            // add to warnings, no need to error
777            warnings.push(WarningReason::ParamNotDeclared(items));
778        }
779
780        if !defined_more_than_once.is_empty() {
781            // sort for consistency
782            let mut items = defined_more_than_once
783                .into_iter()
784                .map(String::from)
785                .collect::<Vec<String>>();
786            items.sort();
787
788            return Err(ParseProvidedParamsError::ParamsProvidedMoreThanOnce(items));
789        }
790
791        let declared_param_names = mpl_params
792            .iter()
793            .filter_map(|p| {
794                // Skip optional params since they don't need to be provided.
795                if p.typ.is_optional() {
796                    None
797                } else {
798                    Some(p.name.as_str())
799                }
800            })
801            .collect::<HashSet<&str>>();
802        let declared_but_not_provided = declared_param_names
803            .difference(&seen)
804            .collect::<Vec<&&str>>();
805        if !declared_but_not_provided.is_empty() {
806            // sort for consistency
807            let mut items = declared_but_not_provided
808                .into_iter()
809                .map(|s| String::from(*s))
810                .collect::<Vec<String>>();
811            items.sort();
812
813            return Err(ParseProvidedParamsError::ParamsDeclaredButNotProvided(
814                items,
815            ));
816        }
817
818        Ok((ProvidedParams::new(provided_params), warnings))
819    }
820
821    /// Return a ref to the inner value.
822    #[must_use]
823    pub fn as_slice(&self) -> &[ProvidedParam] {
824        self.inner.as_slice()
825    }
826
827    fn get_param(&self, name: &str) -> Result<&ProvidedParam, ResolveError> {
828        self.inner
829            .iter()
830            .find(|p| p.name == name)
831            .ok_or(ResolveError::ParamNotProvided(name.to_string()))
832    }
833
834    /// Resolve a `TagValue`.
835    pub fn inline_params(&self, expr: Expr) -> Result<Expr, ResolveError> {
836        let param = match expr {
837            Expr::Const(val) => return Ok(Expr::Const(val)), // no need to resolve
838            Expr::Tag(tag) => return Ok(Expr::Tag(tag)),     // no need to resolve
839            Expr::Param { span: _, param } => param,
840            Expr::Array(parts) => {
841                let parts = parts
842                    .into_iter()
843                    .map(|expr| self.inline_params(expr))
844                    .collect::<Result<_, ResolveError>>()?;
845                return Ok(Expr::Array(parts));
846            }
847            Expr::String(parts) => {
848                // Inline all param expressions in the string concatination
849                let parts = parts
850                    .into_iter()
851                    .map(|part| match part {
852                        StringFragment::Text(text) => Ok(StringFragment::Text(text)),
853                        StringFragment::Expr(expr) => {
854                            Ok(StringFragment::Expr(self.inline_params(expr)?))
855                        }
856                    })
857                    .collect::<Result<Vec<_>, ResolveError>>()?;
858                // If all parts are text, collapse the string
859                return if parts.iter().all(|part| {
860                    matches!(part, StringFragment::Text(_))
861                        | matches!(part, StringFragment::Expr(Expr::Const(_)))
862                }) {
863                    // Collapse the string into a single text fragment,
864                    // there should not be a expr here!
865                    Ok(Expr::Const(
866                        parts
867                            .into_iter()
868                            .map(|part| match part {
869                                StringFragment::Text(text) => text,
870                                // we need to split this out so we avoid the PII safe
871                                // string formating
872                                StringFragment::Expr(Expr::Const(TagValue::String(s))) => {
873                                    s.to_string()
874                                }
875                                StringFragment::Expr(Expr::Const(c)) => c.to_string(),
876                                StringFragment::Expr(_) => {
877                                    "unreachable string collapse".to_string()
878                                }
879                            })
880                            .collect::<String>()
881                            .try_into()?,
882                    ))
883                } else {
884                    Ok(Expr::String(parts))
885                };
886            }
887        };
888
889        let provided_param = self.get_param(&param.name)?;
890        match &provided_param.value {
891            ParamValue::String(val) => {
892                Ok(Expr::Const(TagValue::String(SharedString::try_from(val)?)))
893            }
894            ParamValue::Int(val) => Ok(Expr::Const(TagValue::Int(*val))),
895            ParamValue::Float(val) => Ok(Expr::Const(TagValue::Float(*val))),
896            ParamValue::Bool(val) => Ok(Expr::Const(TagValue::Bool(*val))),
897            ParamValue::Array(val) => Ok(Expr::Const(TagValue::Array(val.clone()))),
898            val => Err(ResolveError::InvalidType {
899                name: param.name,
900                defined: val.typ(),
901                expected: vec![
902                    TerminalParamType::Tag(TagType::String),
903                    TerminalParamType::Tag(TagType::Int),
904                    TerminalParamType::Tag(TagType::Float),
905                    TerminalParamType::Tag(TagType::Bool),
906                ],
907            }),
908        }
909    }
910
911    /// Resolve a `Dataset`.
912    pub fn resolve_dataset(&self, pv: Parameterized<Dataset>) -> Result<Dataset, ResolveError> {
913        let param = match pv {
914            Parameterized::Concrete(val) => return Ok(val), // no need to resolve
915            Parameterized::Param { span: _, param } => param,
916        };
917
918        let provided_param = self.get_param(&param.name)?;
919        match &provided_param.value {
920            ParamValue::Dataset(dataset) => Ok(dataset.clone()),
921            val => Err(ResolveError::InvalidType {
922                name: param.name,
923                defined: val.typ(),
924                expected: vec![TerminalParamType::Dataset],
925            }),
926        }
927    }
928
929    /// Resolve a `RelativeTime`, aka duration.
930    pub fn resolve_relative_time(
931        &self,
932        pv: Parameterized<RelativeTime>,
933    ) -> Result<RelativeTime, ResolveError> {
934        let param = match pv {
935            Parameterized::Concrete(val) => return Ok(val), // no need to resolve
936            Parameterized::Param { span: _, param } => param,
937        };
938
939        let provided_param = self.get_param(&param.name)?;
940        match &provided_param.value {
941            ParamValue::Duration(relative_time) => Ok(relative_time.clone()),
942            val => Err(ResolveError::InvalidType {
943                name: param.name,
944                defined: val.typ(),
945                expected: vec![TerminalParamType::Duration],
946            }),
947        }
948    }
949
950    /// Resolve a regex.
951    pub fn resolve_regex(
952        &self,
953        pv: Parameterized<EncodableRegex>,
954    ) -> Result<EncodableRegex, ResolveError> {
955        let param = match pv {
956            Parameterized::Concrete(val) => return Ok(val), // no need to resolve
957            Parameterized::Param { span: _, param } => param,
958        };
959
960        let provided_param = self.get_param(&param.name)?;
961        match &provided_param.value {
962            ParamValue::Regex(re) => Ok(re.clone()),
963            val => Err(ResolveError::InvalidType {
964                name: param.name,
965                defined: val.typ(),
966                expected: vec![TerminalParamType::Regex],
967            }),
968        }
969    }
970    /// Checks if a param was provided
971    #[must_use]
972    pub fn contains(&self, param: &str) -> bool {
973        self.get_param(param).is_ok()
974    }
975
976    /// Returns the filter when it should be applied for these params.
977    ///
978    /// Plain filters are always active. `ifdef` filters are active only when
979    /// their guarding optional param was provided by the caller.
980    #[must_use]
981    pub fn active_filter<'a>(&self, filter: &'a FilterOrIfDef) -> Option<&'a Filter> {
982        match filter {
983            FilterOrIfDef::Filter(filter) => Some(filter),
984            FilterOrIfDef::Ifdef { param, filter, .. } if self.contains(&param.name) => {
985                Some(filter)
986            }
987            FilterOrIfDef::Ifdef { else_filter, .. } => else_filter.as_ref(),
988        }
989    }
990
991    /// Returns filters that should be applied for these params, preserving order.
992    #[must_use]
993    pub fn active_filters<'a>(&self, filters: &'a [FilterOrIfDef]) -> Vec<&'a Filter> {
994        filters
995            .iter()
996            .filter_map(|filter| self.active_filter(filter))
997            .collect()
998    }
999}
1000
1001/// Parameters that will be set externally.
1002pub type Params = Vec<ParamDeclaration>;
1003
1004/// A Query AST representing a query in the `MPL` language
1005#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1006pub enum Query {
1007    /// A simple query that will produce a result
1008    Simple {
1009        /// The source of the data
1010        source: Source,
1011        /// The filters to apply to the data
1012        filters: Vec<FilterOrIfDef>,
1013        /// The aggregates to apply to the data
1014        aggregates: Vec<Aggregate>,
1015        /// The directives
1016        directives: Directives,
1017        /// The params
1018        params: Params,
1019        /// Tag extends to apply to the series
1020        #[serde(default, skip_serializing_if = "Vec::is_empty")]
1021        extends: Vec<TagExtend>,
1022        /// How to sample series
1023        sample: Option<f64>,
1024    },
1025    /// A compute query taking the input of two queries and producing a by computing combined values
1026    Compute {
1027        /// The left hand side query to compute
1028        left: Box<Query>,
1029        /// The right hand side query to compute
1030        right: Box<Query>,
1031        /// The name of the metric to produce
1032        name: Metric,
1033        /// The compute operation used to combine the left and right queries
1034        op: ComputeFunction,
1035        /// The aggregates to apply to the combined data
1036        aggregates: Vec<Aggregate>,
1037        /// The tag extends to apply to the combined data
1038        #[serde(default, skip_serializing_if = "Vec::is_empty")]
1039        extends: Vec<TagExtend>,
1040        /// The directives
1041        directives: Directives,
1042        /// The params
1043        params: Params,
1044    },
1045}
1046
1047impl Query {
1048    /// Gets the time range for the query
1049    #[must_use]
1050    pub fn time_range(&self) -> Option<&TimeRange> {
1051        match self {
1052            Query::Simple { source, .. } => source.time(),
1053            Query::Compute { left, .. } => left.time_range(),
1054        }
1055    }
1056    /// Get a ref to the params of the query.
1057    #[must_use]
1058    pub fn params(&self) -> &Params {
1059        match self {
1060            Query::Simple { params, .. } | Query::Compute { params, .. } => params,
1061        }
1062    }
1063    /// Get a ref to the directives of the query.
1064    #[must_use]
1065    pub fn directives(&self) -> &Directives {
1066        match self {
1067            Query::Simple { directives, .. } | Query::Compute { directives, .. } => directives,
1068        }
1069    }
1070}
1071
1072impl RelativeTime {
1073    /// Converts a relative time to a `Duration`
1074    pub fn to_duration(&self) -> Result<Duration, TimeError> {
1075        let v = i64::try_from(self.value).map_err(TimeError::InvalidDuration)?;
1076        Ok(match self.unit {
1077            TimeUnit::Millisecond => Duration::milliseconds(v),
1078            TimeUnit::Second => Duration::seconds(v),
1079            TimeUnit::Minute => Duration::minutes(v),
1080            TimeUnit::Hour => Duration::hours(v),
1081            TimeUnit::Day => Duration::days(v),
1082            TimeUnit::Week => Duration::weeks(v),
1083            TimeUnit::Month => Duration::days(v.saturating_mul(30)),
1084            TimeUnit::Year => Duration::days(v.saturating_mul(365)),
1085        })
1086    }
1087
1088    /// Converts a relative time to a `Resolution`
1089    pub fn to_resolution(&self) -> Result<Resolution, ResolutionError> {
1090        match self.unit {
1091            TimeUnit::Millisecond => Resolution::secs(self.value / 1000),
1092            TimeUnit::Second => Resolution::secs(self.value),
1093            TimeUnit::Minute => Resolution::secs(self.value.saturating_mul(60)),
1094            TimeUnit::Hour => Resolution::secs(self.value.saturating_mul(60 * 60)),
1095            TimeUnit::Day => Resolution::secs(self.value.saturating_mul(60 * 60 * 24)),
1096            TimeUnit::Week => Resolution::secs(self.value.saturating_mul(60 * 60 * 24 * 7)),
1097            TimeUnit::Month => Resolution::secs(self.value.saturating_mul(60 * 60 * 24 * 30)),
1098            TimeUnit::Year => Resolution::secs(self.value.saturating_mul(60 * 60 * 24 * 365)),
1099        }
1100    }
1101}
1102
1103/// An error that can occur when converting a time value.
1104#[derive(Debug, thiserror::Error)]
1105pub enum TimeError {
1106    /// Invalid timestamp could not be converted to a UTC datetime
1107    #[error("Invalid timestamp {0}, could not be converted to a UTC datetime")]
1108    InvalidTimestamp(i64),
1109    /// Invalid duration could not be converted to Duration as it exceeds the maximum i64
1110    #[error(
1111        "Invalid duration {0}, could not be converted to Duration as it exceeds the maximum i64"
1112    )]
1113    InvalidDuration(TryFromIntError),
1114}
1115#[cfg(feature = "clock")]
1116impl Time {
1117    fn to_datetime(&self) -> Result<DateTime<Utc>, TimeError> {
1118        Ok(match self {
1119            Time::Relative(t) => Utc::now() - t.to_duration()?,
1120            Time::Timestamp(ts) => {
1121                DateTime::<Utc>::from_timestamp(*ts, 0).ok_or(TimeError::InvalidTimestamp(*ts))?
1122            }
1123            Time::RFC3339(t) => t.with_timezone(&Utc),
1124            Time::Modifier(_) => todo!(),
1125        })
1126    }
1127}
1128
1129#[cfg(feature = "clock")]
1130impl TimeRange {
1131    /// Converts a time range to a start and pair
1132    pub fn to_start_end(&self) -> Result<(DateTime<Utc>, DateTime<Utc>), TimeError> {
1133        let start = self.start.to_datetime()?;
1134        let end = self
1135            .end
1136            .as_ref()
1137            .map_or_else(|| Ok(Utc::now()), Time::to_datetime)?;
1138        Ok((start, end))
1139    }
1140}
mpl_lang/query.rs

mpl_lang/
query.rs
