From 601672527cb139370b35907de8d320deaf5b838b Mon Sep 17 00:00:00 2001
From: Maaz Ahmed <maaz.a@subcom.tech>
Date: Fri, 22 Dec 2023 11:47:05 +0000
Subject: [PATCH] feat: initial functions support in the query builder API
---
Cargo.toml | 1 +
src/query/fns/aggregate.rs | 274 +++++++++++++++++++++++++++++++++++++
src/query/fns/label.rs | 118 ++++++++++++++++
src/query/fns/mod.rs | 62 +++++++++
src/query/fns/rollup.rs | 153 +++++++++++++++++++++
src/query/fns/transform.rs | 49 +++++++
src/query/mod.rs | 1 +
src/query/ops/modifiers.rs | 47 +++++--
tests/fns.rs | 111 +++++++++++++++
9 files changed, 804 insertions(+), 12 deletions(-)
create mode 100644 src/query/fns/aggregate.rs
create mode 100644 src/query/fns/label.rs
create mode 100644 src/query/fns/mod.rs
create mode 100644 src/query/fns/rollup.rs
create mode 100644 src/query/fns/transform.rs
create mode 100644 tests/fns.rs
diff --git a/Cargo.toml b/Cargo.toml
index d227ac3..e921aa1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,6 +17,7 @@ rayon = { version = "1.8.0", optional = true }
[features]
default = []
utils = ["dep:serde", "dep:serde_json", "dep:rayon"]
+metricsql = []
[dev-dependencies]
tokio = { version = "1.34.0", default-features = false, features = ["rt", "rt-multi-thread", "macros"]}
diff --git a/src/query/fns/aggregate.rs b/src/query/fns/aggregate.rs
new file mode 100644
index 0000000..cdf5e74
--- /dev/null
+++ b/src/query/fns/aggregate.rs
@@ -0,0 +1,274 @@
+//! Aggregate query functions (includes PromQL aggregation operators)
+use super::{basic_fn, QryFunc};
+use crate::{
+ query::{
+ ops::{
+ modifiers::{AggrType, Mod},
+ Operable,
+ },
+ IntoQuery,
+ },
+ seal::Sealed,
+};
+use core::fmt;
+use std::fmt::Display;
+
+/// Type that represents an aggregate function, also known as the aggregation operator in PromQL
+///
+/// Unlike other functions, this provides a modifier [`by`](Self::by) and [`without`](Self::without). Only one can be used at once.
+pub struct AggrFunc<'a, F: Fn(&mut fmt::Formatter) -> fmt::Result> {
+ inner: QryFunc<F>,
+ mod_type: Mod<'a, AggrType>,
+}
+
+impl<'a, F: Fn(&mut fmt::Formatter) -> fmt::Result> AggrFunc<'a, F> {
+ pub fn by<I: IntoIterator<Item = &'a str>>(mut self, labels: I) -> Self {
+ self.mod_type = Mod::from((AggrType::By, labels));
+ self
+ }
+
+ pub fn without<I: IntoIterator<Item = &'a str>>(mut self, labels: I) -> Self {
+ self.mod_type = Mod::from((AggrType::Without, labels));
+ self
+ }
+}
+
+impl<'a, F: Fn(&mut fmt::Formatter) -> fmt::Result> From<QryFunc<F>> for AggrFunc<'a, F> {
+ fn from(value: QryFunc<F>) -> Self {
+ AggrFunc {
+ inner: value,
+ mod_type: Default::default(),
+ }
+ }
+}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Sealed for AggrFunc<'_, F> {}
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Operable for AggrFunc<'_, F> {}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Display for AggrFunc<'_, F> {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{qry_fn}{mod_type}",
+ qry_fn = self.inner,
+ mod_type = self.mod_type
+ )
+ }
+}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> IntoQuery for AggrFunc<'_, F> {
+ type Target = String;
+ fn into_query(self) -> Self::Target {
+ self.to_string()
+ }
+}
+
+/// The sum aggregate operator/function
+#[inline]
+pub fn sum<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("sum", vec_expr).into()
+}
+
+/// The min aggregate operator/function
+#[inline]
+pub fn min<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("min", vec_expr).into()
+}
+
+/// The max aggregate operator/function
+#[inline]
+pub fn max<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("max", vec_expr).into()
+}
+
+/// The avg aggregate operator/function
+#[inline]
+pub fn avg<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("avg", vec_expr).into()
+}
+
+/// The group aggregate operator/function
+#[inline]
+pub fn group<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("group", vec_expr).into()
+}
+
+/// The stddev aggregate operator/function
+#[inline]
+pub fn stddev<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("stddev", vec_expr).into()
+}
+
+/// The stdvar aggregate operator/function
+#[inline]
+pub fn stdvar<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("stdvar", vec_expr).into()
+}
+
+/// The count aggregate operator/function
+#[inline]
+pub fn count<'a>(
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("count", vec_expr).into()
+}
+
+/// The count_values aggregate operator/function
+#[inline]
+pub fn count_values<'a>(
+ label: &'_ str,
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result + '_> {
+ QryFunc::new(move |f| write!(f, r#"count_values("{label}", {vec_expr})"#)).into()
+}
+
+/// The topk aggregate operator/function
+#[inline]
+pub fn topk<'a>(
+ k: usize,
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ QryFunc::new(move |f| write!(f, "topk({k}, {vec_expr})")).into()
+}
+
+/// The bottomk aggregate operator/function
+#[inline]
+pub fn bottomk<'a>(
+ k: usize,
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ QryFunc::new(move |f| write!(f, "bottomk({k}, {vec_expr})")).into()
+}
+
+/// The bottomk aggregate operator/function (phi value should always be >= 0 and <= 1)
+#[inline]
+pub fn quantile<'a>(
+ phi: f32,
+ vec_expr: impl Operable + 'static,
+) -> AggrFunc<'a, impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ QryFunc::new(move |f| write!(f, "quantile({phi}, {vec_expr})")).into()
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::query::Metric;
+
+ use super::*;
+
+ #[test]
+ fn by_mod() {
+ let agr = AggrFunc {
+ inner: QryFunc::new(|f| f.write_str("func(x)")),
+ mod_type: Default::default(),
+ };
+ let by_str = agr.by(["label", "label2"]).to_string();
+ assert_eq!(by_str, "func(x) by (label,label2)");
+ }
+
+ #[test]
+ fn without_mod() {
+ let agr = AggrFunc {
+ inner: QryFunc::new(|f| f.write_str("func(x)")),
+ mod_type: Default::default(),
+ };
+ let by_str = agr.without(["label", "label2"]).to_string();
+ assert_eq!(by_str, "func(x) without (label,label2)");
+ }
+
+ #[test]
+ fn aggr_sum() {
+ let vec = Metric::new("test_metric");
+ let sum = sum(vec).to_string();
+ assert_eq!(sum, "sum(test_metric)");
+ }
+
+ #[test]
+ fn aggr_min() {
+ let vec = Metric::new("test_metric");
+ let min = min(vec).to_string();
+ assert_eq!(min, "min(test_metric)");
+ }
+
+ #[test]
+ fn aggr_max() {
+ let vec = Metric::new("test_metric");
+ let max = max(vec).to_string();
+ assert_eq!(max, "max(test_metric)");
+ }
+
+ #[test]
+ fn aggr_avg() {
+ let vec = Metric::new("test_metric");
+ let avg = avg(vec).to_string();
+ assert_eq!(avg, "avg(test_metric)");
+ }
+
+ #[test]
+ fn aggr_group() {
+ let vec = Metric::new("test_metric");
+ let group = group(vec).to_string();
+ assert_eq!(group, "group(test_metric)");
+ }
+
+ #[test]
+ fn aggr_stddev() {
+ let vec = Metric::new("test_metric");
+ let stddev = stddev(vec).to_string();
+ assert_eq!(stddev, "stddev(test_metric)");
+ }
+
+ #[test]
+ fn aggr_stdvar() {
+ let vec = Metric::new("test_metric");
+ let stdvar = stdvar(vec).to_string();
+ assert_eq!(stdvar, "stdvar(test_metric)");
+ }
+
+ #[test]
+ fn aggr_count() {
+ let vec = Metric::new("test_metric");
+ let count = count(vec).to_string();
+ assert_eq!(count, "count(test_metric)");
+ }
+
+ #[test]
+ fn aggr_count_values() {
+ let vec = Metric::new("build_version");
+ let count_values = count_values("version", vec).to_string();
+ assert_eq!(count_values, "count_values(\"version\", build_version)");
+ }
+
+ #[test]
+ fn aggr_topk() {
+ let vec = Metric::new("http_requests_total");
+ let topk = topk(5, vec).to_string();
+ assert_eq!(topk, "topk(5, http_requests_total)");
+ }
+
+ #[test]
+ fn aggr_bottomk() {
+ let vec = Metric::new("http_requests_total");
+ let bottomk = bottomk(5, vec).to_string();
+ assert_eq!(bottomk, "bottomk(5, http_requests_total)");
+ }
+ #[test]
+ fn aggr_quantile() {
+ let vec = Metric::new("http_requests_total");
+ let quantile = quantile(0.1, vec).to_string();
+ assert_eq!(quantile, "quantile(0.1, http_requests_total)");
+ }
+}
diff --git a/src/query/fns/label.rs b/src/query/fns/label.rs
new file mode 100644
index 0000000..2bc3d5b
--- /dev/null
+++ b/src/query/fns/label.rs
@@ -0,0 +1,118 @@
+//! Label manipulation query functions
+
+use core::fmt;
+
+use crate::query::ops::Operable;
+
+use super::QryFunc;
+
+/// MetricsQL's label_map query function
+pub fn mql_label_map<'a>(
+ qry_expr: impl Operable + 'a,
+ label: &'a str,
+ src_dst_pairs: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ QryFunc::new(move |f| {
+ write!(
+ f,
+ "label_map({qry_expr},\"{label}\",\"{pairs}\")",
+ pairs = src_dst_pairs.join("\",\"")
+ )
+ })
+}
+
+fn basic_label_fn<'a>(
+ fn_name: &'static str,
+ qry_expr: impl Operable + 'a,
+ labels: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ QryFunc::new(move |f| {
+ write!(
+ f,
+ "{fn_name}({qry_expr},\"{labels}\")",
+ labels = labels.join("\",\"")
+ )
+ })
+}
+
+/// MetricsQL's sort_by_label query function
+pub fn mql_sort_by_label<'a>(
+ qry_expr: impl Operable + 'a,
+ labels: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ basic_label_fn("sort_by_label", qry_expr, labels)
+}
+
+/// MetricsQL's sort_by_label_desc query function
+pub fn mql_sort_by_label_desc<'a>(
+ qry_expr: impl Operable + 'a,
+ labels: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ basic_label_fn("sort_by_label_desc", qry_expr, labels)
+}
+
+/// MetricsQL's sort_by_label_numeric query function
+pub fn mql_sort_by_label_numeric<'a>(
+ qry_expr: impl Operable + 'a,
+ labels: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ basic_label_fn("sort_by_label_numeric", qry_expr, labels)
+}
+/// MetricsQL's sort_by_label_numeric_desc query function
+pub fn mql_sort_by_label_numeric_desc<'a>(
+ qry_expr: impl Operable + 'a,
+ labels: &'a [&'a str],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + 'a> {
+ basic_label_fn("sort_by_label_numeric_desc", qry_expr, labels)
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::query::Metric;
+
+ use super::*;
+
+ #[test]
+ fn label_label_map() {
+ let query = mql_label_map(
+ Metric::new("metric"),
+ "label",
+ &["val", "dest", "val1", "dest1"],
+ )
+ .to_string();
+
+ assert_eq!(
+ query,
+ r#"label_map(metric,"label","val","dest","val1","dest1")"#
+ );
+ }
+
+ #[test]
+ fn label_sort_by_label() {
+ let query = mql_sort_by_label(Metric::new("metric"), &["label", "label2"]).to_string();
+ assert_eq!(query, r#"sort_by_label(metric,"label","label2")"#);
+ }
+
+ #[test]
+ fn label_sort_by_label_desc() {
+ let query = mql_sort_by_label_desc(Metric::new("metric"), &["label", "label2"]).to_string();
+ assert_eq!(query, r#"sort_by_label_desc(metric,"label","label2")"#);
+ }
+
+ #[test]
+ fn label_sort_by_label_numeric() {
+ let query =
+ mql_sort_by_label_numeric(Metric::new("metric"), &["label", "label2"]).to_string();
+ assert_eq!(query, r#"sort_by_label_numeric(metric,"label","label2")"#);
+ }
+
+ #[test]
+ fn label_sort_by_label_numeric_desc() {
+ let query =
+ mql_sort_by_label_numeric_desc(Metric::new("metric"), &["label", "label2"]).to_string();
+ assert_eq!(
+ query,
+ r#"sort_by_label_numeric_desc(metric,"label","label2")"#
+ );
+ }
+}
diff --git a/src/query/fns/mod.rs b/src/query/fns/mod.rs
new file mode 100644
index 0000000..d7f8cab
--- /dev/null
+++ b/src/query/fns/mod.rs
@@ -0,0 +1,62 @@
+//! Differnt types of Query functions (part of the Query builder API)
+//!
+//! Most of the functions are organized in their respective sub-modules, and the categories are taken from MetricsQL.
+//! Any functions that cannot be categorized, will be placed directly within this module.
+//! Note: PromQL's aggregate operators are provided as functions within the [`aggregate`] sub-module.
+//!
+//! Enable the `metricsql` feature flag to use MetricsQL specific functionality.
+//! All the MetricsQL functions are prefixed by `mql_` to distinguish them from the PromQL functions,
+//! as there is an overlap of functions betweent the two query languages. While MetricsQL is backwards
+//! compatible with PromQL, it still extends the query language in many ways that isn't compatible with PromQL. Therefore, separation of the API will avoid any confusion and bugs.
+//!
+//! Unlike the general query builder API provided by this crate, the fns are provided as actual functions.
+//! The returned types implement the `Display` and `IntoQuery` traits and work just like the types from the other parts of the API.
+use super::{ops::Operable, IntoQuery};
+use crate::seal::Sealed;
+use std::{fmt, fmt::Display};
+
+pub mod aggregate;
+#[cfg(feature = "metricsql")]
+pub mod label;
+pub mod rollup;
+#[cfg(feature = "metricsql")]
+pub mod transform;
+
+/// Type that represents a query function. Meant to be constructed through the query functions provided in the [`fns`](crate::query::fns) module.
+///
+/// This internally holds an actual function (closure), that acts as a type that implements the `Display` trait.
+#[derive(Debug, Clone)]
+pub struct QryFunc<F: Fn(&mut fmt::Formatter) -> fmt::Result> {
+ inner: F,
+}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> QryFunc<F> {
+ fn new(func: F) -> Self {
+ Self { inner: func }
+ }
+}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Display for QryFunc<F> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ (self.inner)(f)
+ }
+}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Sealed for QryFunc<F> {}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> Operable for QryFunc<F> {}
+
+impl<F: Fn(&mut fmt::Formatter) -> fmt::Result> IntoQuery for QryFunc<F> {
+ type Target = String;
+ fn into_query(self) -> Self::Target {
+ self.to_string()
+ }
+}
+
+#[inline]
+fn basic_fn(
+ name: &'static str,
+ expr: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + '_> {
+ QryFunc::new(move |f| write!(f, "{name}({expr})"))
+}
diff --git a/src/query/fns/rollup.rs b/src/query/fns/rollup.rs
new file mode 100644
index 0000000..934e243
--- /dev/null
+++ b/src/query/fns/rollup.rs
@@ -0,0 +1,153 @@
+use core::fmt;
+
+use crate::query::ops::Operable;
+
+use super::{basic_fn, QryFunc};
+
+/// The avg_over_time rollup query function
+#[inline]
+pub fn avg_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("avg_over_time", range_vec)
+}
+
+/// The min_over_time rollup query function
+#[inline]
+pub fn min_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("min_over_time", range_vec)
+}
+
+/// The max_over_time rollup query function
+#[inline]
+pub fn max_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("max_over_time", range_vec)
+}
+
+/// The sum_over_time rollup query function
+#[inline]
+pub fn sum_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("sum_over_time", range_vec)
+}
+
+/// The count_over_time rollup query function
+#[inline]
+pub fn count_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("count_over_time", range_vec)
+}
+
+/// The quantile_over_time rollup query function (phi value must be >= 0 and <= 1)
+#[inline]
+pub fn quantile_over_time(
+ phi: f32,
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ QryFunc::new(move |f| write!(f, "quantile_over_time({phi},{range_vec})"))
+}
+
+/// The stddev_over_time rollup query function
+#[inline]
+pub fn stddev_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("stddev_over_time", range_vec)
+}
+
+/// The stdvar_over_time rollup query function
+#[inline]
+pub fn stdvar_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("stdvar_over_time", range_vec)
+}
+
+/// The present_over_time rollup query function
+#[inline]
+pub fn present_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("present_over_time", range_vec)
+}
+
+/// The last_over_time rollup query function
+#[inline]
+pub fn last_over_time(
+ range_vec: impl Operable + 'static,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ basic_fn("last_over_time", range_vec)
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::query::{Duration, Metric, Unit};
+
+ use super::*;
+
+ fn test_metric() -> Metric<'static> {
+ Metric::new("test_metric").lookback(Duration(15, Unit::Day))
+ }
+
+ #[test]
+ fn rollup_avg() {
+ let string = avg_over_time(test_metric()).to_string();
+ assert_eq!(string, "avg_over_time(test_metric[15d])");
+ }
+
+ #[test]
+ fn rollup_min() {
+ let string = min_over_time(test_metric()).to_string();
+ assert_eq!(string, "min_over_time(test_metric[15d])");
+ }
+
+ #[test]
+ fn rollup_max() {
+ let string = max_over_time(test_metric()).to_string();
+ assert_eq!(string, "max_over_time(test_metric[15d])");
+ }
+ #[test]
+ fn rollup_sum() {
+ let string = sum_over_time(test_metric()).to_string();
+ assert_eq!(string, "sum_over_time(test_metric[15d])");
+ }
+ #[test]
+ fn rollup_count() {
+ let string = count_over_time(test_metric()).to_string();
+ assert_eq!(string, "count_over_time(test_metric[15d])");
+ }
+
+ #[test]
+ fn rollup_quantile() {
+ let string = quantile_over_time(0.1, test_metric()).to_string();
+ assert_eq!(string, "quantile_over_time(0.1,test_metric[15d])");
+ }
+ #[test]
+ fn rollup_stddev() {
+ let string = stddev_over_time(test_metric()).to_string();
+ assert_eq!(string, "stddev_over_time(test_metric[15d])");
+ }
+
+ #[test]
+ fn rollup_stdvar() {
+ let string = stdvar_over_time(test_metric()).to_string();
+ assert_eq!(string, "stdvar_over_time(test_metric[15d])");
+ }
+ #[test]
+ fn rollup_present() {
+ let string = present_over_time(test_metric()).to_string();
+ assert_eq!(string, "present_over_time(test_metric[15d])");
+ }
+
+ #[test]
+ fn rollup_last() {
+ let string = last_over_time(test_metric()).to_string();
+ assert_eq!(string, "last_over_time(test_metric[15d])");
+ }
+}
diff --git a/src/query/fns/transform.rs b/src/query/fns/transform.rs
new file mode 100644
index 0000000..e5838fd
--- /dev/null
+++ b/src/query/fns/transform.rs
@@ -0,0 +1,49 @@
+use core::fmt;
+
+use crate::query::ops::Operable;
+
+use super::QryFunc;
+
+/// MetricsQL `limit_offset` query function
+pub fn mql_limit_offset(
+ limit: usize,
+ offset: usize,
+ qry_expr: impl Operable,
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result> {
+ QryFunc::new(move |f| write!(f, "limit_offset({limit},{offset},{qry_expr})"))
+}
+
+/// MetricsQL `union` query function
+pub fn mql_union(
+ qry_expressions: &[impl Operable],
+) -> QryFunc<impl Fn(&mut fmt::Formatter) -> fmt::Result + '_> {
+ QryFunc::new(|f| {
+ write!(
+ f,
+ "union({queries})",
+ queries = qry_expressions
+ .iter()
+ .map(ToString::to_string)
+ .collect::<Vec<_>>()
+ .join(",")
+ )
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::query::Metric;
+
+ #[test]
+ fn trans_limit_offset() {
+ let query = mql_limit_offset(1, 2, Metric::new("metric")).to_string();
+ assert_eq!(query, "limit_offset(1,2,metric)");
+ }
+
+ #[test]
+ fn trans_union() {
+ let query = mql_union(&[Metric::new("metric"), Metric::new("metric2")]).to_string();
+ assert_eq!(query, "union(metric,metric2)");
+ }
+}
diff --git a/src/query/mod.rs b/src/query/mod.rs
index d6cc713..becd60d 100644
--- a/src/query/mod.rs
+++ b/src/query/mod.rs
@@ -7,6 +7,7 @@ use std::fmt::Display;
use crate::seal::Sealed;
+pub mod fns;
pub mod ops;
// Add default implementation of IntoQuery where the type simply returns itself
diff --git a/src/query/ops/modifiers.rs b/src/query/ops/modifiers.rs
index df5b30e..6cd4096 100644
--- a/src/query/ops/modifiers.rs
+++ b/src/query/ops/modifiers.rs
@@ -6,19 +6,10 @@ pub(crate) struct Mod<'a, M> {
labels: Vec<&'a str>,
}
-impl<'a, T: IntoIterator<Item = &'a str>> From<(MatchType, T)> for Mod<'a, MatchType> {
- fn from((match_type, labels): (MatchType, T)) -> Self {
+impl<'a, T: IntoIterator<Item = &'a str>, M> From<(M, T)> for Mod<'a, M> {
+ fn from((mod_type, labels): (M, T)) -> Self {
Mod {
- mod_type: match_type,
- labels: labels.into_iter().collect(),
- }
- }
-}
-
-impl<'a, T: IntoIterator<Item = &'a str>> From<(GroupType, T)> for Mod<'a, GroupType> {
- fn from((grp_type, labels): (GroupType, T)) -> Self {
- Mod {
- mod_type: grp_type,
+ mod_type,
labels: labels.into_iter().collect(),
}
}
@@ -91,3 +82,35 @@ impl Display for GroupType {
})
}
}
+
+#[derive(Default, Debug, Clone, Copy)]
+pub(crate) enum AggrType {
+ By,
+ Without,
+ #[default]
+ None,
+}
+
+impl Display for AggrType {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.write_str(match self {
+ AggrType::By => "by",
+ AggrType::Without => "without",
+ AggrType::None => "",
+ })
+ }
+}
+
+impl Display for Mod<'_, AggrType> {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ if !self.labels.is_empty() {
+ return write!(
+ f,
+ " {mod_type} ({labels})",
+ mod_type = self.mod_type,
+ labels = self.labels.join(",")
+ );
+ }
+ Ok(())
+ }
+}
diff --git a/tests/fns.rs b/tests/fns.rs
new file mode 100644
index 0000000..15c2521
--- /dev/null
+++ b/tests/fns.rs
@@ -0,0 +1,111 @@
+mod utils;
+use mquery::query::{
+ fns::{aggregate, label, rollup, transform},
+ Duration, Metric, Unit,
+};
+
+macro_rules! rollup_tests {
+ ($($fn_name:ident),+) => {
+ $(#[tokio::test]
+ async fn $fn_name() {
+ let query = rollup::$fn_name(Metric::new("test_metric").lookback(Duration(1, Unit::Day)));
+ utils::send_query(query).await.unwrap();
+ })+
+ };
+}
+
+macro_rules! aggr_tests {
+ ($($fn_name: tt),+) => {
+ $(#[tokio::test]
+ async fn $fn_name() {
+ let query = aggregate::$fn_name(Metric::new("test_metric"));
+ utils::send_query(query).await.unwrap();
+ })+
+ };
+}
+
+macro_rules! label_tests {
+ ($($fn_name: tt),+) => {
+ $(#[tokio::test]
+ async fn $fn_name() {
+ let query = label::$fn_name(Metric::new("metric"), &["label", "label2"]);
+ utils::send_query(query).await.unwrap();
+ })+
+ };
+}
+
+aggr_tests!(sum, min, max, avg, group, stddev, stdvar, count);
+
+#[tokio::test]
+async fn aggr_count_values() {
+ let query = aggregate::count_values("label", Metric::new("test_metric"));
+ utils::send_query(query).await.unwrap();
+}
+
+#[tokio::test]
+async fn aggr_topk() {
+ let query = aggregate::topk(10, Metric::new("test_metric"));
+ utils::send_query(query).await.unwrap();
+}
+
+#[tokio::test]
+async fn aggr_bottomk() {
+ let query = aggregate::bottomk(10, Metric::new("test_metric"));
+ utils::send_query(query).await.unwrap();
+}
+
+#[tokio::test]
+async fn aggr_quantile() {
+ let query = aggregate::quantile(0.5, Metric::new("test_metric"));
+ utils::send_query(query).await.unwrap();
+}
+
+rollup_tests!(
+ avg_over_time,
+ min_over_time,
+ max_over_time,
+ sum_over_time,
+ count_over_time,
+ stddev_over_time,
+ stdvar_over_time,
+ present_over_time,
+ last_over_time
+);
+
+#[tokio::test]
+async fn quantile_over_time() {
+ let query = rollup::quantile_over_time(
+ 0.1,
+ Metric::new("test_metric").lookback(Duration(1, Unit::Day)),
+ );
+ utils::send_query(query).await.unwrap();
+}
+
+#[tokio::test]
+async fn mql_label_map() {
+ let query = label::mql_label_map(
+ Metric::new("metric"),
+ "label",
+ &["source", "destination", "source2", "destination2"],
+ );
+ utils::send_query(query).await.unwrap();
+}
+
+label_tests!(
+ mql_sort_by_label,
+ mql_sort_by_label_desc,
+ mql_sort_by_label_numeric,
+ mql_sort_by_label_numeric_desc
+);
+
+#[tokio::test]
+async fn trans_limit_offset() {
+ let query = transform::mql_limit_offset(1, 2, Metric::new("metric")).to_string();
+ utils::send_query(query).await.unwrap();
+}
+
+#[tokio::test]
+async fn trans_union() {
+ let query = transform::mql_union(&[Metric::new("metric"), Metric::new("metric2")]).to_string();
+ utils::send_query(query).await.unwrap();
+}
--
GitLab