Skip to main content

Live series

Everything else in these pages works on a batch: a whole series you already have. A live series takes events as they arrive (from a socket, a queue, a sensor) and keeps whatever is built on it up to date: a filtered view, a running per-minute average. It is the Rust version of pond-ts's LiveSeries, and each push gives the same results as pond-ts, in the same order (checked against cases pond-ts writes out).

The live layer is complete. It covers the live series itself (pushing, ordering rules, how many events to keep, lookups); the views filter, map, select, sample and window; the running ones diff, rate, pct_change, fill and cumulative; aggregate into fixed-length or calendar buckets; rolling windows and reduce, with triggers that set when they emit; and partitions, one series per host (or any column) with views and windows on each.

It lives in pond_rs::live, behind the live cargo feature, which is on by default.

Making one and pushing events​

A live series has a name and a schema: the key column first (time for moments), then the value columns (numbers, text, true/false or arrays). push(time, cells) adds one event: the time in epoch milliseconds and one cell per value column, None for a gap. push_many adds a batch; push_json takes rows written as pond-ts JSON; push_series takes a batch TimeSeries (data that already comes in columns: an Arrow chunk, a JSON read) and is the fastest way in.

use pond_rs::live::LiveSeries;
use pond_rs::{ColumnDef, ColumnKind};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;

    live.push(0, [Some("a".into()), Some(0.25.into())])?;
    live.push_many([
        (1_000, [Some("b".into()), Some(0.5.into())]),
        (2_000, [Some("a".into()), None]),
    ])?;
    live.push_json(r#"[[3000, "b", 0.75]]"#)?;

    assert_eq!(live.len(), 4);
    assert_eq!(live.last().unwrap().number("load")?, Some(0.75));
    assert_eq!(live.at(-2).unwrap().number("load")?, None); // the gap
    assert_eq!(live.at_or_before(1_500).unwrap().string("host")?, Some("b"));

    // A copy of what it holds now, as an ordinary batch series.
    let batch = live.to_time_series()?;
    assert_eq!(batch.len(), 4);
    Ok(())
}

The lookups are the batch ones: at (negative counts from the end), first / last, bisect, at_or_before / at_or_after, includes_key, find / some / every, and events() to walk them all. Events are stored column by column, so each read builds the event it hands you: at, last, events() and the rest give you your own copy (a LiveEvent), not a reference into the series. time_range() is the span from the first event to the last, event_rate() the events per second over it, and stats() the counters (events taken, evicted and skipped). to_json writes what it holds as pond-ts JSON, and LiveSeries::from_json starts one from it.

A batch is all or nothing. push_many checks every row first, then the order of the whole batch; if anything is wrong it refuses the batch and stores none of it. (pond-ts used to keep the rows before a bad one in some modes; that was fixed there to match.)

Pushing a batch series. push_series(&series) adds every row of a TimeSeries as one batch, with the same rules and outputs as push_many of the same rows. The series must be keyed the same way (a time key for a time-keyed live series) and have exactly the live series' value columns by name, in any order, each of the same kind; a gap in a required column is refused like a bad row, naming the row. When nothing is built on the series and nothing watches it, the columns are copied straight into its storage without building an event per row, which is what makes it fast: one million rows in batches of 1,000 take about half the time of push_many (and about a third less than pond-ts's pushMany; see Speed below). pond-ts's pushSeries takes the same batch under the same rules, but keeps the batch's arrays instead of copying them, so it hasn't been timed against push_series. One difference: on a series with partitions, pond-rs refuses a batch with a row for an undeclared group before storing anything, where pond-ts stores the batch and then throws.

use pond_rs::live::LiveSeries;
use pond_rs::{Column, ColumnDef, ColumnKind, Float64Column, StringColumn, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let chunk = TimeSeries::new(
        "chunk",
        vec![0, 1_000, 2_000],
        vec![
            Column::optional("load", Float64Column::from_options([Some(0.25), Some(0.5), None])),
            Column::optional("host", StringColumn::from_options([Some("a"), Some("b"), Some("a")])),
        ],
    )?;
    live.push_series(&chunk)?;
    assert_eq!(live.len(), 3);
    assert_eq!(live.at(1).unwrap().string("host")?, Some("b"));
    Ok(())
}

Late events and how many to keep​

LiveSeries::with_options(name, schema, LiveOptions::new()...) sets two things.

What to do with an event that starts before the latest one (.ordering(…)):

  • OrderingMode::Strict, the default: refuse it with an error.
  • OrderingMode::Drop: skip it quietly, and count it in stats().rejected.
  • OrderingMode::Reorder: store it in its place in time. .grace_window("5s") limits how late it may be; an older one is refused. (To keep every late event but let buckets wait for them, see Per-host buckets over late events.)

How many to keep (oldest go first, after each push):

  • .max_events(n): at most n events.
  • .max_age("1h"): only events that start within that long of the latest one. Age is measured from the latest event's time, not a clock, so a series that stops receiving events keeps what it has.
use pond_rs::live::{LiveOptions, LiveSeries, OrderingMode};
use pond_rs::{ColumnDef, ColumnKind};

fn main() -> Result<(), pond_rs::PondError> {
    let schema = || {
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
        ]
    };

    // Strict: a late event is an error, and nothing changes.
    let mut strict = LiveSeries::new("cpu", schema())?;
    strict.push(5_000, [Some(1.0.into())])?;
    assert!(strict.push(4_000, [Some(2.0.into())]).is_err());
    assert_eq!(strict.len(), 1);

    // Drop: skipped and counted.
    let mut drop = LiveSeries::with_options(
        "cpu",
        schema(),
        LiveOptions::new().ordering(OrderingMode::Drop),
    )?;
    drop.push(5_000, [Some(1.0.into())])?;
    drop.push(4_000, [Some(2.0.into())])?;
    assert_eq!((drop.len(), drop.stats().rejected), (1, 1));

    // Reorder within 2 seconds, keeping the last 3 events.
    let mut reorder = LiveSeries::with_options(
        "cpu",
        schema(),
        LiveOptions::new()
            .ordering(OrderingMode::Reorder)
            .grace_window("2s")
            .max_events(3),
    )?;
    reorder.push(5_000, [Some(1.0.into())])?;
    reorder.push(4_000, [Some(2.0.into())])?; // 1 s late: put in place
    assert!(reorder.push(1_000, [Some(3.0.into())]).is_err()); // 4 s late
    reorder.push(6_000, [Some(4.0.into())])?;
    reorder.push(7_000, [Some(5.0.into())])?; // the 4-second event goes
    let times: Vec<i64> = reorder.events().map(|e| e.begin()).collect();
    assert_eq!(times, [5_000, 6_000, 7_000]);
    Ok(())
}

Building on it: filter and aggregate​

live.filter(source, test) makes a view of the events that pass a test; live.aggregate(source, &Sequence::every("1m")?, &aggs) makes fixed buckets with one column per Agg, as the batch aggregate does. Both start from what the source holds already, then keep up as events arrive. The source is live.root() or another node, so they chain.

The live series owns every node built on it. Building one gives back a small handle (a Node): copyable, safe to send to another thread, and tied to no borrow. Read the node through the series that owns it: live[busy], or live.get(busy)? to get an error instead of a panic for a handle that was removed.

A view drops the events its source drops (by max_events, max_age or clear). An aggregation keeps its buckets whatever the source drops.

  • live[node].len(), .events(), .at(…): the closed buckets, oldest first.
  • .has_open_bucket(): whether a bucket is still collecting.
  • .closed(): the closed buckets as a batch series; .snapshot(): the closed ones plus the open ones as they stand now.

A bucket closes when an event arrives at or after its end. Under OrderingMode::Reorder with a grace window, it stays open until the latest event is a grace window past its end, so late events can still land in it. Calendar buckets (days in a time zone) are below.

use pond_rs::live::LiveSeries;
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer, Sequence};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let busy = live.filter(live.root(), |e| {
        e.number("load").ok().flatten() > Some(0.8)
    })?;
    let per_minute = live.aggregate(
        live.root(),
        &Sequence::every("1m")?,
        &[
            Agg::new("load_avg", "load", Reducer::Avg),
            Agg::new("events", "load", Reducer::Count),
        ],
    )?;

    live.push(0, [Some("a".into()), Some(0.9.into())])?;
    live.push(30_000, [Some("b".into()), Some(0.5.into())])?;
    assert_eq!(live[busy].len(), 1);
    assert_eq!(live[per_minute].len(), 0); // the first minute is still open
    assert!(live[per_minute].has_open_bucket());

    // An event in the second minute closes the first.
    live.push(61_000, [Some("a".into()), Some(0.7.into())])?;
    let minute = live[per_minute].at(0).unwrap();
    assert_eq!(minute.begin(), 0);
    assert_eq!(minute.number("load_avg")?, Some(0.7));
    assert_eq!(minute.number("events")?, Some(2.0));

    // The closed minute, plus the open one as it stands.
    assert_eq!(live[per_minute].closed()?.len(), 1);
    assert_eq!(live[per_minute].snapshot()?.len(), 2);
    Ok(())
}

Aggregating a text, true/false or array column works with count, first, last, keep, unique, top<N> and samples. Number reducers (sum, avg, …) on those columns are an error; pond-ts gives 0 or gaps instead (logged as a pond-ts issue, [PND-KINDREDUCE]).

Calendar buckets and a grace of its own​

aggregate also takes calendar buckets: Sequence::calendar("day", "America/New_York") makes one bucket per day in that time zone, so the day the clocks move is 23 or 25 hours long, as in the batch aggregate. Days, weeks, months, quarters and years all work. While a calendar aggregation exists, the series refuses times outside the years −9997 to 9997; pond-ts takes any time.

By default an aggregation built on the series waits for the series' grace window before closing a bucket. aggregate_with(from, &sequence, &aggs, Some("10m")) gives it its own grace instead (Some("0s") closes a bucket as soon as an event lands past its end), as pond-ts's new LiveAggregation(source, sequence, mapping, { grace }). An aggregation built on a view has no grace unless you give it one.

use pond_rs::live::{LiveOptions, LiveSeries, OrderingMode};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer, Sequence};

fn main() -> Result<(), pond_rs::PondError> {
    let hour = 3_600_000;
    // Midnight UTC on 9, 10 and 11 March 2024. New York's clocks moved
    // forward on the 10th, so that day there is 23 hours long.
    let (mar9, mar10, mar11) = (1_709_942_400_000, 1_710_028_800_000, 1_710_115_200_000);

    let mut live = LiveSeries::with_options(
        "sales",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("amount", ColumnKind::Number),
        ],
        // Late events up to 2 hours are put in place.
        LiveOptions::new().ordering(OrderingMode::Reorder).grace_window("2h"),
    )?;
    let new_york_day = Sequence::calendar("day", "America/New_York")?;
    let total = [Agg::new("total", "amount", Reducer::Sum)];
    // Takes the series' 2-hour grace: a day stays open 2 hours past its end.
    let daily = live.aggregate(live.root(), &new_york_day, &total)?;
    // Its own grace of 0: a day closes as soon as the next one starts.
    let daily_now = live.aggregate_with(live.root(), &new_york_day, &total, Some("0s"))?;

    live.push(mar9 + 17 * hour, [Some(10.0.into())])?; // noon on the 9th, New York
    live.push(mar10 + 16 * hour, [Some(20.0.into())])?; // noon on the 10th
    live.push(mar11 + 5 * hour, [Some(30.0.into())])?; // 1 am on the 11th
    // 11 pm on the 10th, arriving 2 hours late.
    live.push(mar11 + 3 * hour, [Some(5.0.into())])?;
    // 4 am on the 11th: now 2 hours past the end of the 10th.
    live.push(mar11 + 8 * hour, [Some(1.0.into())])?;

    let day = live[daily].at(1).unwrap();
    // The 10th runs from 05:00 UTC (midnight EST) to 04:00 UTC the next
    // day (midnight EDT): 23 hours.
    assert_eq!(day.begin(), mar10 + 5 * hour);
    assert_eq!(day.end() - day.begin(), 23 * hour);
    assert_eq!(day.number("total")?, Some(25.0)); // the late 5 counted

    // With no grace, the 10th had closed before the late event came.
    let day = live[daily_now].at(1).unwrap();
    assert_eq!(day.number("total")?, Some(20.0));
    Ok(())
}

Clock triggers on rolling windows and reduce still take even steps only (Trigger::every), as in pond-ts.

More views​

Every view takes a source (live.root() or another view), starts from what that source holds, and then keeps up. Chain them freely: a fill on the root, a rate on the fill, a window on the rate.

Reshaping and trimming:

  • map(from, |event| cells): new values for each event, one cell per value column in schema order. The time and the columns stay the same.
  • select(from, &["load"]): only the named columns.
  • sample(from, Sample::Stride(n)): every nth event, counted from when the view is built.
  • window(from, "5m"): only events within that long of the newest one. window_rows(from, n): only the newest n. live[view].event_rate() gives events per second over a time window.
use pond_rs::live::LiveSeries;
use pond_rs::{ColumnDef, ColumnKind, Sample};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
            ColumnDef::optional("mem", ColumnKind::Number),
        ],
    )?;
    let percent = live.map(live.root(), |e| {
        let load = e.number("load").ok().flatten();
        vec![load.map(|l| (l * 100.0).into()), e.value("mem").unwrap().cloned()]
    })?;
    let load_only = live.select(live.root(), &["load"])?;
    let every_other = live.sample(live.root(), Sample::Stride(2))?;
    let last_2s = live.window(live.root(), "2s")?;
    let last_two = live.window_rows(live.root(), 2)?;

    for (i, load) in [0.25, 0.5, 0.75, 1.0, 0.5].into_iter().enumerate() {
        live.push(i as i64 * 1_000, [Some(load.into()), Some(64.0.into())])?;
    }

    assert_eq!(live[percent].last().unwrap().number("load")?, Some(50.0));
    assert_eq!(live[load_only].schema().len(), 2); // time and load
    let sampled: Vec<i64> = live[every_other].events().map(|e| e.begin()).collect();
    assert_eq!(sampled, [1_000, 3_000]);
    assert_eq!(live[last_2s].len(), 3); // 2 s back from 4 s: from 2 s on
    assert_eq!(live[last_2s].event_rate()?, 1.5); // 3 events in 2 seconds
    assert_eq!(live[last_two].len(), 2);
    Ok(())
}

Running calculations, one event at a time, on the columns you name (other columns pass through):

  • diff(from, &["bytes"]): the change since the previous event. rate(…): that change per second. pct_change(…): as a fraction of the previous value. The first event gets a gap; the _with forms take FirstRow::Drop to leave it out.
  • fill(from, &[("bytes", Fill::Hold)]): fill gaps as events arrive, with Fill::Hold (the last value seen), Fill::Zero or Fill::Value(…). fill_all(from, Fill::Hold) fills every column. Fill::Bfill and Fill::Linear need events that have not arrived yet, so live views refuse them, as pond-ts does.
  • cumulative(from, &[("bytes", Reducer::Sum)]): a running sum, Max, Min or Count. A gap keeps the running value. cumulative_with(from, "bytes", |acc, v| …) runs a fold of your own.
use pond_rs::live::LiveSeries;
use pond_rs::{ColumnDef, ColumnKind, Fill, FirstRow, Reducer, Value};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "net",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("bytes", ColumnKind::Number),
        ],
    )?;
    // Fill the missing reading, then take bytes per second of the filled values.
    let filled = live.fill(live.root(), &[("bytes", Fill::Hold)])?;
    let per_second = live.rate(filled, &["bytes"])?;
    let changes = live.diff_with(live.root(), &["bytes"], FirstRow::Drop)?;
    let total = live.cumulative(live.root(), &[("bytes", Reducer::Sum)])?;

    for (t, bytes) in [(0, Some(100.0)), (1_000, None), (2_000, Some(300.0)), (4_000, Some(600.0))] {
        live.push(t, [bytes.map(Value::from)])?;
    }

    let filled: Vec<_> = live[filled].events().map(|e| e.number("bytes").unwrap()).collect();
    assert_eq!(filled, [Some(100.0), Some(100.0), Some(300.0), Some(600.0)]);
    let rate: Vec<_> = live[per_second].events().map(|e| e.number("bytes").unwrap()).collect();
    assert_eq!(rate, [None, Some(0.0), Some(200.0), Some(150.0)]);
    // A gap next to either side gives a gap; the first event is left out.
    let diffs: Vec<_> = live[changes].events().map(|e| e.number("bytes").unwrap()).collect();
    assert_eq!(diffs, [None, None, Some(300.0)]);
    let sums: Vec<_> = live[total].events().map(|e| e.number("bytes").unwrap()).collect();
    assert_eq!(sums, [Some(100.0), Some(100.0), Some(400.0), Some(1000.0)]);
    Ok(())
}

A few things to know:

  • Mistakes are refused when the view is built, not ignored: an unknown column, diff or cumulative on a column that is not numbers, a fill value of the wrong kind, a stride or row window of 0. pond-ts takes most of these silently and gives gaps.
  • fill_with(…, limit) fills at most limit gaps in a row. A live view cannot know how long a run of gaps will be, so it fills the first limit and leaves the rest; the batch fill_with leaves a too-long run unfilled.
  • A window trims itself quietly. It returns no Evict output when events fall out of it, and views built on it keep what it trimmed (as in pond-ts). To drop old events everywhere, set max_events or max_age on the series instead.
  • Late events under Reorder go into every view at their place in time. The running views (diff, rate, fill, cumulative, sample) work in arrival order, so a late event's change is against the event that arrived just before it, not the one just before it in time. pond-ts used to throw an error here and now does the same.

Rolling windows and reduce​

Two nodes keep running results instead of passing events on:

  • rolling(from, "30s", &aggs) works out each Agg over a trailing window: the events starting no more than 30 s before the newest one (an event exactly 30 s back still counts). By default it emits one event per source event, keyed at that event.
  • rolling_rows(from, n, &aggs, &options) does the same over the last n events instead of a span of time.
  • rolling_windows(from, &[RollingWindow::new("1m", &aggs), …], &options) works out several time windows in one pass over one shared buffer and emits them together, one column per Agg across all windows (output names must differ). It is cheaper than one rolling per window.
  • reduce(from, &aggs) works out each Agg over everything from holds now: values are added as events arrive and taken out again as from drops them (by max_events, max_age, clear, or a window view trimming itself). It emits once at the end of each push, keyed at the push's last event, however many rows the push brought.

rolling takes the series, a view or an aggregation; reduce takes the series or a view. Nothing can be built on a rolling window or a reduce (pond-ts has no methods there either). Every reducer the batch aggregate has works here, including sums of huge numbers kept exact.

RollingOptions sets the rest, for rolling_with, rolling_rows and rolling_windows:

  • .min_samples(n): emit gaps until the window holds n events (gaps count). RollingWindow::min_samples sets it for one fused window.
  • .trigger(…): when to emit. Trigger::event() (the default) for every event; Trigger::count(n) every nth event; Trigger::every("1m") when an event crosses into a new minute, keyed at that minute's start (the first event only sets the minute; a quiet minute emits nothing). reduce_with(from, &aggs, trigger) takes the same triggers.
  • .history(…): how many emitted events to keep for reading back: History::All (the default), History::Off, or History::Limit with a max_events and/or a max_age. Watchers get every emitted event whatever this says.

Read the current result with live[node].value() (one value per column, in order, without emitting), the kept events with .events() / .at(i) (at(-1) is the latest), and counters with .stats(). Watching a rolling window's OutputKind::Update gives its values after every source event, even when its trigger did not fire.

use pond_rs::live::{History, LiveSeries, RollingOptions, RollingWindow, Trigger};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    // The average and peak over the last 2 seconds, after every event.
    let last_2s = live.rolling(
        live.root(),
        "2s",
        &[
            Agg::new("avg", "load", Reducer::Avg),
            Agg::new("peak", "load", Reducer::Max),
        ],
    )?;
    // The average of the last 3 events, once 3 have arrived, every 2nd
    // event, keeping only the latest result.
    let last_3 = live.rolling_rows(
        live.root(),
        3,
        &[Agg::new("avg", "load", Reducer::Avg)],
        &RollingOptions::new()
            .min_samples(3)
            .trigger(Trigger::count(2)?)
            .history(History::Limit { max_events: Some(1), max_age: None }),
    )?;
    // Two windows in one pass, emitted together.
    let both = live.rolling_windows(
        live.root(),
        &[
            RollingWindow::new("1s", &[Agg::new("avg_1s", "load", Reducer::Avg)]),
            RollingWindow::new("4s", &[Agg::new("avg_4s", "load", Reducer::Avg)]),
        ],
        &RollingOptions::new(),
    )?;

    for (i, load) in [1.0, 2.0, 3.0, 4.0, 5.0].into_iter().enumerate() {
        live.push(i as i64 * 1_000, [Some(load.into())])?;
    }

    // At 4 s the window reaches back to 2 s: 3, 4 and 5.
    let latest = live[last_2s].at(-1).unwrap();
    assert_eq!(latest.begin(), 4_000);
    assert_eq!(latest.number("avg")?, Some(4.0));
    assert_eq!(latest.number("peak")?, Some(5.0));
    assert_eq!(live[last_2s].len(), 5);
    assert_eq!(live[last_2s].value(), [Some(4.0.into()), Some(5.0.into())]);

    // Emitted at the 2nd and 4th events (the 2nd still warming up: a gap);
    // only the last is kept.
    assert_eq!(live[last_3].len(), 1);
    let kept = live[last_3].at(0).unwrap();
    assert_eq!((kept.begin(), kept.number("avg")?), (3_000, Some(3.0)));
    assert_eq!(live[last_3].stats().emissions, 2);

    let row = live[both].at(-1).unwrap();
    assert_eq!(row.number("avg_1s")?, Some(4.5));
    assert_eq!(row.number("avg_4s")?, Some(3.0));
    Ok(())
}

A reduce over the series follows what the series keeps, so with max_events or max_age it is a total over a moving set of events:

use pond_rs::live::{LiveOptions, LiveSeries, Output, OutputKind, Trigger};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::with_options(
        "orders",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("amount", ColumnKind::Number),
        ],
        LiveOptions::new().max_events(3),
    )?;
    // Totals over whatever the series holds: its last 3 orders.
    let totals = live.reduce(
        live.root(),
        &[
            Agg::new("total", "amount", Reducer::Sum),
            Agg::new("orders", "amount", Reducer::Count),
        ],
    )?;
    // The same, but only once per minute.
    let per_minute = live.reduce_with(
        live.root(),
        &[Agg::new("total", "amount", Reducer::Sum)],
        Trigger::every("1m")?,
    )?;
    live.watch(totals, OutputKind::Event)?;

    // One push of three rows: one emission, keyed at its last row.
    let out = live.push_many(vec![
        (0, vec![Some(10.0.into())]),
        (20_000, vec![Some(20.0.into())]),
        (40_000, vec![Some(30.0.into())]),
    ])?;
    assert_eq!(out.len(), 1);
    let Output::Event { event, .. } = &out[0] else { panic!() };
    assert_eq!(event.begin(), 40_000);
    assert_eq!(event.number("total")?, Some(60.0));

    // A fourth order pushes the first out of the series, and out of the total.
    live.push(65_000, [Some(5.0.into())])?;
    let now = live[totals].at(-1).unwrap();
    assert_eq!(now.number("total")?, Some(55.0));
    assert_eq!(now.number("orders")?, Some(3.0));

    // The per-minute reduce emitted once, when 65 s crossed into minute 1,
    // keyed at the minute's start.
    assert_eq!(live[per_minute].len(), 1);
    let minute = live[per_minute].at(0).unwrap();
    assert_eq!((minute.begin(), minute.number("total")?), (60_000, Some(55.0)));
    Ok(())
}

Partitions: one series per host​

Events from many sources often arrive mixed together: readings from a thousand hosts, trades in many symbols. Running a diff or a rolling average over the mixed stream compares one host with another. partition_by(live.root(), "host", &options) splits the series into one series per value of a column, built when the value first arrives, so each host's events get their own views and windows.

  • PartitionOptions: each partition takes the root's ordering, grace window and retention unless you set them (.ordering, .grace_window, .max_events, .max_age; setting either retention limit replaces both). .groups(&["a", "b"]) declares the values up front: their partitions are built at once, and a push with any other value is refused.
  • live[hosts] reads the partitioned node: .keys() lists the values in the order their partitions were built, .partition("a") gives that host's series (read it like the root: len, events, at, last…), and .partitions() gives every value with its series.
  • per_partition(hosts, |live, host| …) describes a chain of nodes to build on every partition: the closure gets one partition, builds on it with the usual methods (diff, fill, rolling, …) and returns the last node's id. Nothing is built yet. The closure runs again for every partition, including hosts that appear later, so it must build the same chain each time. It also runs once straight away on an empty series, to learn the output columns and catch mistakes.
  • collect(from, &CollectOptions::new()) gathers into one series everything the partitions take, or everything a per_partition chain emits. It only gathers: what a partition drops stays. apply(hosts, |live, host| …, &options) is per_partition and collect in one call.
  • partition_outputs(chain) builds the chain on each partition there is now and returns each value with its output's id; live.node::<LiveView>(id)? turns an id into a handle you can read.
use pond_rs::live::{CollectOptions, LiveSeries, LiveView, PartitionOptions};
use pond_rs::{ColumnDef, ColumnKind};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "net",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("bytes", ColumnKind::Number),
        ],
    )?;
    // One series per host, each keeping its last 100 events.
    let hosts = live.partition_by(live.root(), "host", &PartitionOptions::new().max_events(100))?;
    // The change in bytes since the same host's previous event: a diff
    // built on every host's series, including hosts that appear later.
    let per_host_diff =
        live.per_partition(hosts, |live, host| Ok(live.diff(host, &["bytes"])?.id()))?;
    // Every host's diffs gathered back into one series.
    let changes = live.collect(per_host_diff, &CollectOptions::new())?;

    live.push(0, [Some("a".into()), Some(100.0.into())])?;
    live.push(1_000, [Some("b".into()), Some(500.0.into())])?;
    live.push(2_000, [Some("a".into()), Some(160.0.into())])?;
    live.push(3_000, [Some("b".into()), Some(520.0.into())])?;

    assert_eq!(live[hosts].keys(), ["a", "b"]);
    let a = live[hosts].partition("a").unwrap();
    assert_eq!(live[a].len(), 2);

    // Each change is against the same host's previous value, not the
    // previous event overall.
    let diffs: Vec<_> = live[changes]
        .events()
        .map(|e| (e.begin(), e.string("host").unwrap().map(String::from), e.number("bytes").unwrap()))
        .collect();
    assert_eq!(
        diffs,
        [
            (0, Some("a".into()), None),
            (1_000, Some("b".into()), None),
            (2_000, Some("a".into()), Some(60.0)),
            (3_000, Some("b".into()), Some(20.0)),
        ]
    );

    // Or build the chain on each host there is now, and read one host's.
    let built = live.partition_outputs(per_host_diff)?;
    let (key, id) = &built[1];
    let b_diff = live.node::<LiveView>(*id)?;
    assert_eq!(key, "b");
    assert_eq!(live[b_diff].last().unwrap().number("bytes")?, Some(20.0));
    Ok(())
}

Every host on one clock. partition_rolling(hosts, "30s", &aggs, &options) keeps a rolling window per host and, each time the clock in Trigger::every passes a mark, emits one row for every host seen so far, keyed at the mark: the time, the host, then the columns. A quiet host's window is first brought up to date, so it reports what its last 30 seconds hold at that moment. partition_rolling_rows and partition_rolling_windows do the same with a count window or several time windows. These need a clock trigger; for a window that emits on every event, build rolling with per_partition.

use pond_rs::live::{LiveSeries, PartitionOptions, RollingOptions, Trigger};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let hosts = live.partition_by(live.root(), "host", &PartitionOptions::new())?;
    // Each host's average over its last 30 seconds, every host reported
    // together each time the clock passes a 10-second mark.
    let every_10s = live.partition_rolling(
        hosts,
        "30s",
        &[Agg::new("load_avg", "load", Reducer::Avg)],
        &RollingOptions::new().trigger(Trigger::every("10s")?),
    )?;

    live.push(0, [Some("a".into()), Some(1.0.into())])?;
    live.push(5_000, [Some("b".into()), Some(3.0.into())])?;
    live.push(10_000, [Some("a".into()), Some(2.0.into())])?; // passes 10 s

    // One row per host, keyed at the 10-second mark.
    let rows: Vec<_> = live[every_10s]
        .events()
        .map(|e| {
            let host = e.string("host").unwrap().unwrap().to_string();
            (e.begin(), host, e.number("load_avg").unwrap())
        })
        .collect();
    assert_eq!(
        rows,
        [(10_000, "a".to_string(), Some(1.5)), (10_000, "b".to_string(), Some(3.0))]
    );
    Ok(())
}

Per-host buckets over late events​

A common shape: readings from many hosts arrive out of order, and each host needs its own minute buckets that wait a few seconds for stragglers before closing. Three things trip people up here:

  • Put the delay on the buckets, not the series. A series' grace window refuses any event later than it, and the whole push fails. To keep every late event but hold buckets open for a while, give the series no grace window and pass the delay to aggregate_with.
  • Carry the host into the buckets. A closed bucket has only its time span and its reducer columns, and the series apply gathers mixes every host. A first reducer on host puts it back. (partition_outputs names each partition, but only the ones that exist when it is called.)
  • Buckets from different hosts close out of time order, so the gathered series needs Reorder. It takes the partitions' ordering unless you set one.
use pond_rs::live::{CollectOptions, LiveOptions, LiveSeries, OrderingMode, PartitionOptions};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer, Sequence};

fn main() -> Result<(), pond_rs::PondError> {
    // Readings arrive out of order. No grace window on the series, so no
    // late event is refused; the 5-second delay goes on the buckets.
    let mut live = LiveSeries::with_options(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("host", ColumnKind::String),
            ColumnDef::optional("cpu", ColumnKind::Number),
        ],
        LiveOptions::new().ordering(OrderingMode::Reorder),
    )?;
    let hosts = live.partition_by(live.root(), "host", &PartitionOptions::new())?;
    // Minute buckets per host. A bucket closes once its own host's latest
    // event is 5 s past its end. The `host` column carries the host into
    // the gathered series, which otherwise mixes every host's buckets.
    let minutes = live.apply(
        hosts,
        |live, host| {
            let buckets = live.aggregate_with(
                host,
                &Sequence::every("1m")?,
                &[
                    Agg::new("host", "host", Reducer::First),
                    Agg::new("cpu_avg", "cpu", Reducer::Avg),
                    Agg::new("readings", "cpu", Reducer::Count),
                ],
                Some("5s"),
            )?;
            Ok(buckets.id())
        },
        // Buckets from different hosts can close out of time order. The
        // gathered series takes the partitions' ordering (Reorder here);
        // setting it says so.
        &CollectOptions::new().ordering(OrderingMode::Reorder),
    )?;

    let s = 1_000;
    live.push(10 * s, [Some("a".into()), Some(1.0.into())])?;
    live.push(20 * s, [Some("b".into()), Some(5.0.into())])?;
    live.push(50 * s, [Some("a".into()), Some(3.0.into())])?;
    live.push(62 * s, [Some("a".into()), Some(2.0.into())])?;
    live.push(59 * s, [Some("a".into()), Some(5.0.into())])?; // late, still in time
    live.push(66 * s, [Some("a".into()), Some(1.0.into())])?; // closes a's first minute
    live.push(58 * s, [Some("a".into()), Some(100.0.into())])?; // too late: left out
    assert_eq!(live[minutes].len(), 1);
    live.push(70 * s, [Some("b".into()), Some(7.0.into())])?; // closes b's first minute

    let closed: Vec<_> = live[minutes]
        .events()
        .map(|e| {
            let host = e.string("host").unwrap().unwrap().to_string();
            (host, e.number("cpu_avg").unwrap(), e.number("readings").unwrap())
        })
        .collect();
    assert_eq!(
        closed,
        [("a".to_string(), Some(3.0), Some(3.0)), ("b".to_string(), Some(5.0), Some(1.0))]
    );
    Ok(())
}

What a push returns: outputs​

In pond-ts you attach listeners (series.on('event', fn)) and pond-ts calls them. pond-rs calls none of your code during a push (apart from a filter's test). Instead, each push returns what it caused, as a list of Outputs, for the nodes and kinds you asked for with live.watch(node, kind):

KindComes fromMeans
OutputKind::Eventthe series, a viewIt took an event
OutputKind::Eventan aggregationA bucket closed (the event is the finished bucket)
OutputKind::Batchthe seriesAll the events one push stored
OutputKind::Evictthe series, a viewEvents it dropped (retention or clear)
OutputKind::Bucketan aggregationA bucket as it stands after an event joined it
OutputKind::Updatean aggregationIt took an event, even one too late for any bucket
OutputKind::Eventa rolling windowIt emitted (per its trigger)
OutputKind::Updatea rolling windowIt took an event; carries its values as they stand
OutputKind::Eventa reduceIt emitted (at the end of a push)

Nothing is returned for what you don't watch, so an unwatched node costs nothing extra. The outputs come in the order pond-ts's listeners would fire: each event travels down the tree (a node's subscribers in the order they were built), then the series' own retention, Batch and Evict. output.node() says which node it came from; compare it with handle.id(). live.unwatch(node, kind) stops it.

use pond_rs::live::{LiveSeries, Output, OutputKind};
use pond_rs::{Agg, ColumnDef, ColumnKind, Reducer, Sequence};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let per_minute = live.aggregate(
        live.root(),
        &Sequence::every("1m")?,
        &[Agg::new("load_avg", "load", Reducer::Avg)],
    )?;
    live.watch(per_minute, OutputKind::Event)?;
    live.watch(per_minute, OutputKind::Bucket)?;

    // Joining an open bucket: just the bucket as it stands.
    let out = live.push(0, [Some(1.0.into())])?;
    assert_eq!(out.len(), 1);
    assert!(matches!(&out[0], Output::Bucket { event, .. } if event.begin() == 0));

    // Starting the next minute: the new bucket first, then the closed one
    // (pond-ts's order).
    let out = live.push(60_000, [Some(3.0.into())])?;
    let kinds: Vec<OutputKind> = out.iter().map(|o| o.kind()).collect();
    assert_eq!(kinds, [OutputKind::Bucket, OutputKind::Event]);
    let Output::Event { node, event } = &out[1] else { panic!() };
    assert_eq!(*node, per_minute.id());
    assert_eq!((event.begin(), event.number("load_avg")?), (0, Some(1.0)));
    Ok(())
}

Sending outputs somewhere else​

Because a push only returns a list, where the outputs go next is up to you: a channel, a tokio::sync::watch (when only the latest value matters), a websocket, or a plain function call. A channel takes a few lines:

use pond_rs::live::{LiveSeries, Output, OutputKind};
use pond_rs::{ColumnDef, ColumnKind};
use std::sync::mpsc;
use std::thread;

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let busy = live.filter(live.root(), |e| {
        e.number("load").ok().flatten() > Some(0.8)
    })?;
    live.watch(busy, OutputKind::Event)?;

    let (tx, rx) = mpsc::channel::<Output>();
    let alerts = thread::spawn(move || {
        // Another thread reads the alerts as they come.
        rx.iter()
            .filter_map(|o| match o {
                Output::Event { event, .. } => Some(event.begin()),
                _ => None,
            })
            .collect::<Vec<i64>>()
    });

    for (t, load) in [(0, 0.5), (1_000, 0.9), (2_000, 0.95), (3_000, 0.1)] {
        for output in live.push(t, [Some(load.into())])? {
            tx.send(output).unwrap();
        }
    }
    drop(tx); // no more alerts: the reader's loop ends
    assert_eq!(alerts.join().unwrap(), [1_000, 2_000]);
    Ok(())
}

Threads. A LiveSeries can be moved to another thread. To read it from several threads, put it behind a lock (Arc<Mutex<LiveSeries>>), or publish a to_time_series() copy after each push for readers to use.

Removing a node​

live.remove(node) takes a node out (pond-ts's dispose). Its handle then gives an error from live.get(…). Nodes built on it keep what they hold but get no more events. As in pond-ts, when you remove a view, the view it was built on is detached too if nothing else uses it (no other node, nothing watched); it stays readable but stops updating. The series itself cannot be removed.

use pond_rs::live::LiveSeries;
use pond_rs::{ColumnDef, ColumnKind};

fn main() -> Result<(), pond_rs::PondError> {
    let mut live = LiveSeries::new(
        "cpu",
        vec![
            ColumnDef::required("time", ColumnKind::Time),
            ColumnDef::optional("load", ColumnKind::Number),
        ],
    )?;
    let busy = live.filter(live.root(), |e| {
        e.number("load").ok().flatten() > Some(0.8)
    })?;
    live.push(0, [Some(0.9.into())])?;

    live.remove(busy)?;
    assert!(live.get(busy).is_err());
    live.push(1_000, [Some(0.95.into())])?; // the series still takes events
    assert_eq!(live.len(), 2);
    Ok(())
}

What differs from pond-ts​

  • Outputs instead of listeners, as above. pond-ts calls your listeners during the push; pond-rs returns the same outputs, in the same order, after it.
  • Stricter schemas. Exact decimal columns and two columns with the same name are refused; so are number reducers on text columns (pond-ts answers those with 0 or gaps).
  • keep on an array column compares the arrays' contents, as the batch aggregate does. pond-ts compares the array objects, so two equal arrays read as different there ([PND-ARRAYKEY]).
  • Stricter view arguments, refused when the view is built (see More views).
  • map cannot change an event's time, only its values; pond-ts's can.
  • Reading a rolling window's value() changes nothing. In pond-ts a read can move the last bit of a later stdev. pond-rs makes the same reads pond-ts makes during a push, so pushed results still match bit for bit.
  • A reduce built over events its source already holds emits for them straight away, into its kept events but no push's outputs. pond-ts emits on a microtask just after the build, so a caller that pushes again without yielding gets one emission covering both from pond-ts, and two from pond-rs.
  • A clock trigger on a source keyed by ranges or intervals is refused; pond-ts emits time keys under the interval schema there.
  • Calendar aggregations take years −9997 to 9997 only, as above.
  • Partitions are built on the root only; pond-ts can partition any series.
  • A push a partition would refuse is refused whole, before anything is stored: a value outside the declared groups, or an event a partition's ordering refuses (as after a clear, which empties the root but not the partitions). pond-ts stores it in the root, delivers the rest and then throws.
  • Removing a partitioned node also removes the per_partition chains described on it.
  • The host column of partition_rolling output is text (the value as pond-ts writes it, " undefined" for a row without one). pond-ts declares it with the source column's type but also fills it with text.

Speed​

One million events, pushed one at a time unless noted, timed on one 4-core Linux machine on 2026-10-29 (ms; pond-rs is the middle of three runs, pond-ts one run with each case in its own process; repeat runs differ by up to a third). Cases marked 10k keep only the last 10,000 events (max_events).

Casepond-rspond-tsFaster
Push only3313,94412×
Push only, 10k2732,84610×
Reorder, 5% of events up to 1 s late3487302.1×
Batches of 1,000322131pond-ts 2.5× faster
A filter and a 1-minute aggregate4594,78310×
The same, every output watched8635,5226.4×
The same, 10k5734,7708.3×
diff + rate + pct_change, 10k1,2277,2695.9×
fill (hold) + cumulative, 10k7355,9498.1×
map + select + sample, 10k8736,1527.0×
Two windows, 10k5294,8959.2×
Reorder (5% late), diff + fill + window, 10k1,5191,8941.2×
rolling 30 s: avg + max + stdev, 10k5505,4109.8×
rolling_rows 100: avg, history off, 10k4885,16511×
rolling_windows 1 m + 5 m + 15 m: avg, 10k4975,97012×
reduce: avg + max + count, 10k63411,09017×
aggregate by New York day, 10k3274,68314×
4 hosts, collect, 10k8528,53210×
4 hosts, diff + fill on each, collect, 10k2,02910,4115.1×
4 hosts, partition_rolling_windows 1 m + 5 m, every 1 m, 10k8957,2748.1×
1,000 hosts in batches of 1,000, rolling on each, 10k1,2503,2382.6×
1,000 hosts in batches of 1,000, partition_rolling, 10k2,7767,4502.7×

So one series takes about 3 million events a second in pond-rs, against about 250,000 in pond-ts, and each view or rolling window costs pond-rs a few tenths of a microsecond per event. Read the big ratios with three things in mind:

  • pond-ts's one-event push is slow in the default mode. A series in the default (in order) mode stores events in column chunks built for batches, and a one-event push builds a whole one-row chunk (about 60% of the time goes to the garbage collector). In a separate check, one million in-order events pushed one at a time into a Reorder series took pond-ts 0.76–0.8 s rather than 3.9–4.4 s, and the late-events case above takes it 730 ms. Against that path pond-rs is about 2× faster, not 12×.
  • pond-ts is faster on batches. In batches of 1,000 pond-ts appends whole column chunks. pond-rs now stores a batch's cells straight from the rows when nothing reads the series (about 255 ms per million rows after the 2026-10-29 speed review, from 322), and most of what is left is freeing the caller's one Vec per row and the host text, which node does in its garbage collector, off the clock.
  • Reorder with views is close (1.2× here, 1.2–1.7× across the three pond-rs runs, the noisiest case). pond-ts's Reorder path skips the one-row chunk above, so its Reorder cases are its fastest.

These timings found several pond-ts slowdowns along the way, fixed in pond-ts at the same time: dropping old events copied the whole buffer on every push (the 10k diff case took 115 s before the fix); pushing one at a time got slower as the series grew; and a loop of pushes that never waited queued one task per push behind a reduce.

Threads and memory allocators​

A LiveSeries and everything built on it runs on the thread that pushes. On a 4-core Linux machine, one thread handled 50 hosts at about 1.0–1.4 million rows a second (pushes of 1,000 to 100,000 rows, a 1 s rolling average per host, with collect or partition_rolling; about 1 µs per row in all). Splitting partitions over threads inside one series was tried and was slower, not faster: each row's work per host is too small to be worth handing to another core. For more than one core, run several series, each taking a share of the hosts; on 1,000 hosts that ran 2.5× faster on 4 threads.

A faster memory allocator helps more than threads. Pushes make and free many small allocations, and swapping the system allocator (glibc, on Linux) for mimalloc made the 50-host case 30–45% faster, with no other change (add the crate, then #[global_allocator] static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; in your program).

Memory​

A live series stores each column as its own list: a start time per event, numbers as plain 8-byte floats, text and true/false as optional values. Bytes held per event, one million events:

ColumnsBefore #842Now
Time + 3 numbers24934
Time + text + 2 numbers25051
Time + 10 numbers58592
Time + text + 2 numbers, plus a filter that keeps everything (series and view together)266102

One thing to plan for: a view keeps its own copy of the events it holds rather than sharing the series', so a view that keeps everything costs about as much memory as the series itself (before #842 such a view cost about 16 bytes per event, against 51 now). Put window, max_events or max_age where you can.