Struct GroupSpec

Source

pub struct GroupSpec { /* private fields */ }

Expand description

Groups of rows with equal key values, and aggregates over each group.

use std::sync::Arc;
use arrow_array::{ArrayRef, Decimal128Array, Int64Array, RecordBatch, StringArray};
use densa_data::{Aggregate, GroupSpec, LiveTable};

let trades = |ids: Vec<i64>, desks: Vec<&str>, qty: Vec<i64>| {
    RecordBatch::try_from_iter([
        ("id", Arc::new(Int64Array::from(ids)) as ArrayRef),
        ("desk", Arc::new(StringArray::from(desks)) as ArrayRef),
        ("qty", Arc::new(Int64Array::from(qty)) as ArrayRef),
    ])
};
let mut table = LiveTable::new(trades(vec![], vec![], vec![])?.schema(), "id")?;
let by_desk = table.add_groups(
    GroupSpec::new()
        .group_by("desk")
        .aggregate(Aggregate::count())
        .aggregate(Aggregate::sum("qty")),
)?;
table.upsert(trades(vec![1, 2, 3], vec!["fx", "rates", "fx"], vec![5, 7, 10])?)?;
table.upsert(trades(vec![3], vec!["rates"], vec![10])?)?; // trade 3 moves desk

let totals = table.groups(by_desk).unwrap().to_batch()?; // fx: 1 trade, rates: 2
let qty = totals.column(2).as_any().downcast_ref::<Decimal128Array>().unwrap();
assert_eq!((qty.value(0), qty.value(1)), (5, 17));

Implementations§

Source §

impl GroupSpec

Source

pub fn new() -> Self

No keys and no aggregates yet. Without keys there is exactly one group, of every counted row, even when there are none.

Source

pub fn group_by(self, column: impl Into<String>) -> Self

Groups by column too, after the key columns given before it. Groups are listed by key in the order views sort: ascending, nulls last. -0.0 and 0.0 are one key, and so are all NaNs.

Source

pub fn group_by_time(self, column: impl Into<String>, every: Duration) -> Self

Groups by buckets of every (a minute, an hour) of column, a column of timestamps, dates, times of day or durations, as a key after the ones added before it.

The buckets are back to back from the column’s zero (midnight UTC for timestamps, whatever their time zone), so a bucket of a minute starts on a whole minute; each value is in the bucket that starts at or before it, less than every earlier. Buckets have one length: a week of dates starts on a Thursday, like 1970-01-01, and there are no months. The key column holds each bucket’s start, in the column’s type and under its name, so a column is a key once (a second key of it is DataError::DuplicateName). The lowest bucket of a type’s range, which may start before it, starts at the type’s smallest value. every is taken in whole units of the column, rounding down, and must be at least one (DataError::BucketWidth otherwise). Null times make a group of their own, last.

With a Window on the same column, the oldest bucket holds only the rows the window still keeps, and goes once they expire.

use std::sync::Arc;
use std::time::Duration;
use arrow_array::{ArrayRef, Int64Array, RecordBatch, TimestampSecondArray, UInt64Array};
use densa_data::{Aggregate, GroupSpec, LiveTable};

let requests = |ids: Vec<i64>, times: Vec<i64>| {
    RecordBatch::try_from_iter([
        ("id", Arc::new(Int64Array::from(ids)) as ArrayRef),
        ("time", Arc::new(TimestampSecondArray::from(times)) as ArrayRef),
    ])
};
let mut table = LiveTable::new(requests(vec![], vec![])?.schema(), "id")?;
let per_minute = table.add_groups(
    GroupSpec::new()
        .group_by_time("time", Duration::from_secs(60))
        .aggregate(Aggregate::count()),
)?;
table.upsert(requests(vec![1, 2, 3], vec![5, 59, 61])?)?;

let counts = table.groups(per_minute).unwrap().to_batch()?;
let starts = counts.column(0).as_any().downcast_ref::<TimestampSecondArray>().unwrap();
let count = counts.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
assert_eq!(starts.values(), &[0, 60]); // 00:00 and 00:01
assert_eq!(count.values(), &[2, 1]);

Source

pub fn group_by_step(self, column: impl Into<String>, step: f64) -> Self

Groups by buckets of step of column, a column of integers, decimals or floats, as a key after the ones added before it. step is in the column’s own units: 0.5 is fifty cents of a decimal of dollars.

As with group_by_time, the buckets are back to back from zero, each value is in the one that starts at or before it, and the key column holds each bucket’s start (the type’s smallest or largest value for a bucket that starts past the type’s range). For integers and decimals step is taken in whole units, rounding down, but a step within a billionth of its size of a whole number counts as it (0.29 is 29 cents, though 0.29 * 100.0 is just below 29); it must be at least one unit, or above zero and finite for floats (DataError::BucketWidth otherwise).

Float buckets forgive the rounding of the column’s type: a value at most two roundings below a bucket’s start, and never more than a sixteenth of a step, is in that bucket. 0.3 is in the bucket of 0.1 steps that starts at 0.3 (3.0 * 0.1, itself just above 0.3), though 0.3 / 0.1 is just below 3. NaN, the infinities and nulls are groups of their own.

Source