Repository: rcoh/angle-grinder
Pull request #142 (2021-07-07 06:20:39)

[lang] add timeslice() operator
This change is a first pass at adding the timeslice() operator.  It seems to do the job, most of the work is done by the chrono crate's duration_trunc() method.

Related to #10



Patch:
diff --git a/README.md b/README.md
index 74365e5..059b853 100644
--- a/README.md
+++ b/README.md
@@ -12,7 +12,7 @@ Angle grinder can process well above 1M rows per second (simple pipelines as hig
 * [Query Syntax Overview](#query-syntax)
 * [Operators](#operators)
     * Parsers: [JSON](#json) [logfmt](#logfmt) [split](#split) [generic](#parse)
-    * Misc: [Add/remove fields](#fields) [limit](#limit) [where](#where)
+    * Misc: [Add/remove fields](#fields) [limit](#limit) [timeslice](#timeslice) [where](#where)
     * Aggregators: [count](#count) [sum](#sum) [min](#min) [max](#max) [percentile](#percentile) [sort](#sort) [total](#total) [count distinct](#count-distinct)
 * [Output Control](#rendering)
 ## Installation
@@ -364,6 +364,27 @@ In addition to columns, `sort` can also sort an arbitrary expressions.
 * | json | sort by length(endpoint_url)
 ```
 
+##### Timeslice
+`timeslice(<timestamp>) <duration> [as <field>]`: Truncates a timestamp to the
+given duration to allow for partitioning messages into slices of time.  The
+`timestamp` parameter must be a date value, such as that returned by the
+`parseDate()` function.  The duration is an amount followed by one of the
+following units:
+
+* `s` - seconds
+* `m` - minutes
+* `h` - hours
+* `d` - days
+* `w` - weeks
+
+The resulting timestamp is placed in the `_timeslice` field by default or the
+field specified after the `as` keyword.
+
+*Examples*:
+```agrind
+* | json | timeslice(parseDate(ts)) 5m
+```
+
 ##### Total
 `total(a) [as renamed_total]`: Compute the running total of a given field. Total does not currently support grouping!
 
diff --git a/src/lang.rs b/src/lang.rs
index 0cb910f..fb2baf8 100644
--- a/src/lang.rs
+++ b/src/lang.rs
@@ -69,7 +69,15 @@ pub const VALID_AGGREGATES: &[&str] = &[
 ];
 
 pub const VALID_INLINE: &[&str] = &[
-    "parse", "limit", "json", "logfmt", "total", "fields", "where", "split",
+    "parse",
+    "limit",
+    "json",
+    "logfmt",
+    "total",
+    "fields",
+    "where",
+    "split",
+    "timeslice",
 ];
 
 lazy_static! {
@@ -281,6 +289,11 @@ pub enum InlineOperator {
         input_column: Option<Expr>,
         output_column: Option<Expr>,
     },
+    Timeslice {
+        input_column: Expr,
+        duration: Option<chrono::Duration>,
+        output_column: Option<String>,
+    },
     Total {
         input_column: Expr,
         output_column: String,
@@ -384,6 +397,18 @@ named!(value<Span, data::Value>, ws!(
     )
 ));
 
+named!(duration<Span, chrono::Duration>, ws!(do_parse!(
+    amount: map!(digit1, |s| s.fragment.0.parse::<i64>().unwrap()) >>
+    value: alt!(
+        map!(tag!("s"), |_| chrono::Duration::seconds(amount)) |
+        map!(tag!("m"), |_| chrono::Duration::minutes(amount)) |
+        map!(tag!("h"), |_| chrono::Duration::hours(amount)) |
+        map!(tag!("d"), |_| chrono::Duration::days(amount)) |
+        map!(tag!("w"), |_| chrono::Duration::weeks(amount))
+    ) >>
+    (value)
+)));
+
 named!(dot_property<Span, DataAccessAtom>, do_parse!(
     field: preceded!(tag!("."), ident) >>
     (DataAccessAtom::Key(field))
@@ -560,6 +585,18 @@ named!(limit<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
     })
 ))));
 
+named!(timeslice<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
+    tag!("timeslice") >>
+    input_column: delimited!(tag!("("), expr, tag!(")")) >>
+    duration: opt!(duration) >>
+    output_column: opt!(ws!(preceded!(tag!("as"), ident))) >>
+    (InlineOperator::Timeslice {
+        input_column,
+        duration,
+        output_column
+    })
+))));
+
 named!(split<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
     tag!("split") >>
     from_column_opt: opt!(delimited!(
@@ -764,7 +801,7 @@ named!(alias<Span, Operator>, map!(
 
 named!(inline_operator<Span, Operator>, do_parse!(
     peek!(did_you_mean_operator) >>
-    res: map!(alt_complete!(parse | json | logfmt | fields | whre | limit | total | split), Operator::Inline) >>
+    res: map!(alt_complete!(parse | json | logfmt | fields | whre | limit | total | split | timeslice), Operator::Inline) >>
     (res)
 ));
 
@@ -1263,6 +1300,32 @@ mod tests {
                 },
             })
         );
+        expect!(
+            operator,
+            r#" timeslice(date) 5m "#,
+            Operator::Inline(Positioned {
+                start_pos: QueryPosition(1),
+                end_pos: QueryPosition(20),
+                value: InlineOperator::Timeslice {
+                    input_column: Expr::column("date"),
+                    duration: Some(chrono::Duration::minutes(5)),
+                    output_column: None,
+                },
+            })
+        );
+        expect!(
+            operator,
+            r#" timeslice(date) 5m as theslice "#,
+            Operator::Inline(Positioned {
+                start_pos: QueryPosition(1),
+                end_pos: QueryPosition(32),
+                value: InlineOperator::Timeslice {
+                    input_column: Expr::column("date"),
+                    duration: Some(chrono::Duration::minutes(5)),
+                    output_column: Some("theslice".to_string()),
+                },
+            })
+        );
     }
 
     #[test]
diff --git a/src/operator.rs b/src/operator.rs
index e354207..3edb1da 100644
--- a/src/operator.rs
+++ b/src/operator.rs
@@ -1,6 +1,7 @@
 use crate::data;
 use crate::data::{Aggregate, DisplayConfig, Record, Row};
 use crate::funcs;
+use chrono::DurationRound;
 use failure::Fail;
 use failure::_core::fmt::Debug;
 use quantiles::ckms::CKMS;
@@ -40,6 +41,12 @@ pub enum EvalError {
     #[fail(display = "Expected number, found {}", found)]
     ExpectedNumber { found: String },
 
+    #[fail(
+        display = "Expected date, found '{}'.  Use parseDate() to convert a value to a date",
+        found
+    )]
+    ExpectedDate { found: String },
+
     #[fail(display = "Expected positive number, found {}", found)]
     ExpectedPositiveNumber { found: String },
 
@@ -68,6 +75,9 @@ pub enum EvalError {
 
     #[fail(display = "The '{}' function failed with the error: {}", name, msg)]
     FunctionFailed { name: &'static str, msg: String },
+
+    #[fail(display = "Duration is not valid for timeslice: {}", error)]
+    InvalidDuration { error: String },
 }
 
 pub trait Evaluatable<T>: Send + Sync + Clone {
@@ -988,6 +998,54 @@ impl UnaryPreAggFunction for FieldExpressionDef {
     }
 }
 
+#[derive(Clone)]
+pub struct Timeslice {
+    input_column: Expr,
+    duration: chrono::Duration,
+    output_column: Option<String>,
+}
+
+impl Timeslice {
+    pub fn new(
+        input_column: Expr,
+        duration: chrono::Duration,
+        output_column: Option<String>,
+    ) -> Self {
+        Self {
+            input_column,
+            duration,
+            output_column,
+        }
+    }
+}
+
+impl UnaryPreAggFunction for Timeslice {
+    fn process(&self, rec: Record) -> Result<Option<Record>, EvalError> {
+        let inp: data::Value = self.input_column.eval(&rec.data)?;
+
+        match inp {
+            data::Value::DateTime(dt) => {
+                let rounded =
+                    dt.duration_trunc(self.duration)
+                        .map_err(|e| EvalError::InvalidDuration {
+                            error: format!("{:?}", e),
+                        })?;
+                let rec = rec.put(
+                    self.output_column
+                        .clone()
+                        .unwrap_or_else(|| "_timeslice".to_string()),
+                    data::Value::DateTime(rounded),
+                );
+
+                Ok(Some(rec))
+            }
+            _ => Err(EvalError::ExpectedDate {
+                found: format!("{}", inp.to_string()),
+            }),
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct Split {
     separator: String,
diff --git a/src/typecheck.rs b/src/typecheck.rs
index 89a12bc..9f86306 100644
--- a/src/typecheck.rs
+++ b/src/typecheck.rs
@@ -27,6 +27,9 @@ pub enum TypeError {
 
     #[fail(display = "Unknown function {}", name)]
     UnknownFunction { name: String },
+
+    #[fail(display = "Expected a duration for the timeslice (e.g. 1h)")]
+    ExpectedDuration,
 }
 
 pub trait TypeCheck<O> {
@@ -279,6 +282,18 @@ impl TypeCheck<Box<dyn operator::OperatorBuilder + Send + Sync>>
                     .map(|e| e.type_check(error_builder))
                     .transpose()?,
             ))),
+            lang::InlineOperator::Timeslice { duration: None, .. } => {
+                Err(TypeError::ExpectedDuration)
+            }
+            lang::InlineOperator::Timeslice {
+                input_column,
+                duration: Some(duration),
+                output_column,
+            } => Ok(Box::new(operator::Timeslice::new(
+                input_column.type_check(error_builder)?,
+                duration,
+                output_column,
+            ))),
             lang::InlineOperator::Total {
                 input_column,
                 output_column,


Test patch:
diff --git a/tests/structured_tests/timeslice_1.toml b/tests/structured_tests/timeslice_1.toml
new file mode 100644
index 0000000..31818fd
--- /dev/null
+++ b/tests/structured_tests/timeslice_1.toml
@@ -0,0 +1,25 @@
+query = """* | json | timeslice(parseDate(ts)) 5m | count by _timeslice"""
+input = """
+{"ts": "2013-09-06T20:00:48.124817Z", "lvl": "TRACE", "msg": "trace test"}
+{"ts": "2013-09-06T20:00:49.124817Z", "lvl": "INFO", "msg": "Starting up service"}
+{"ts": "2013-09-06T22:00:49.124817Z", "lvl": "INFO", "msg": "Shutting down service", "user": "steve@example.com"}
+{"ts": "2013-09-06T22:00:59.124817Z", "lvl": "DEBUG5", "msg": "Details..."}
+{"ts": "2013-09-06T22:00:59.124817Z", "lvl": "DEBUG4", "msg": "Details..."}
+{"ts": "2013-09-06T22:00:59.124817Z", "lvl": "DEBUG3", "msg": "Details..."}
+{"ts": "2013-09-06T22:00:59.124817Z", "lvl": "DEBUG2", "msg": "Details..."}
+{"ts": "2013-09-06T22:00:59.124817Z", "lvl": "DEBUG", "msg": "Details..."}
+{"ts": "2013-09-06T22:01:49.124817Z", "lvl": "STATS", "msg": "1 beat per second"}
+{"ts": "2013-09-06T22:05:49.124817Z", "lvl": "WARNING", "msg": "not looking good"}
+{"ts": "2013-09-06T22:05:49.124817Z", "lvl": "ERROR", "msg": "looking bad"}
+{"ts": "2013-09-06T22:10:49.124817Z", "lvl": "CRITICAL", "msg": "sooo bad"}
+{"ts": "2013-09-06T23:05:49.124817Z", "lvl": "FATAL", "msg": "shoot", "obj": { "field1" : "hi", "field2": 2 }, "arr" : ["hi", {"sub1": true}]}
+"""
+output = """
+_timeslice                     _count
+---------------------------------------------
+2013-09-06 22:00:00 UTC        7
+2013-09-06 20:00:00 UTC        2
+2013-09-06 22:05:00 UTC        2
+2013-09-06 22:10:00 UTC        1
+2013-09-06 23:05:00 UTC        1
+"""
diff --git a/tests/structured_tests/timeslice_2.toml b/tests/structured_tests/timeslice_2.toml
new file mode 100644
index 0000000..a00c87c
--- /dev/null
+++ b/tests/structured_tests/timeslice_2.toml
@@ -0,0 +1,9 @@
+query = """* | json | timeslice(ts)"""
+input = """
+{"ts": "2013-09-06T20:00:48.124817Z", "lvl": "TRACE", "msg": "trace test"}
+"""
+output = """"""
+error = """
+Error: Expected a duration for the timeslice (e.g. 1h)
+"""
+succeeds = false