Rolling Windows

Sliding-window computations over ordered events. For each event in the source, look at a window of nearby events and compute a reducer.

TypeScript

const rollingAvg = cpu.rolling('5m', { cpu: 'avg' });
// rollingAvg.length === cpu.length; each output's cpu is the
// trailing-5-minute average ending at that event.

Rust

use pond_rs::{Agg, Column, Reducer, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    // One reading a minute for ten minutes.
    let cpu = TimeSeries::new(
        "cpu",
        (0..10).map(|i| i * 60_000).collect(),
        vec![Column::required(
            "cpu",
            vec![0.3, 0.5, 0.4, 0.9, 0.6, 0.2, 0.7, 0.8, 0.5, 0.4],
        )],
    )?;

    let rolling_avg = cpu.rolling("5m", &[Agg::of("cpu", Reducer::Avg)])?;
    // One row per source row; each cpu is the trailing-5-minute average
    // ending at that row.
    assert_eq!(rolling_avg.len(), cpu.len());
    let avg = rolling_avg.column("cpu")?.numbers()?;
    assert_eq!(avg.get(4), Some(0.54)); // (0.3 + 0.5 + 0.4 + 0.9 + 0.6) / 5
    assert_eq!(avg.get(9), Some(0.52)); // minutes 5 to 9
    Ok(())
}

Agg::of("cpu", Reducer::Avg) is the shorthand { cpu: 'avg' }: the output keeps the source's name.

Same reducer vocabulary as aggregate — every built-in works, plus custom functions. The difference is per-event output cadence vs. one-per-bucket. For the mental model placing rolling alongside aggregate, reduce, and the streaming variant, see Concepts → Windowing.

Window alignment

What "nearby" means is the alignment. Pick the one that matches the question:

Raw samples (blue); a sliding window (orange) anchored at the anchor (the marker) reduces the samples inside it, tracing the rolling line (teal, the real rolling output). anchor picks what the window rides on — each event or a regular sequence; alignment places the window trailing / centered / leading around the anchor. (See @pond-ts/charts for how the chart is built.)

• 'trailing' (default) — "what's the average of what I've seen up to now."
• 'centered' — "smoothed around now."
• 'leading' — "what's the average of what's coming next."

TypeScript

cpu.rolling('5m', { cpu: 'avg' }, { alignment: 'centered' });

Rust

use pond_rs::{Agg, Alignment, Column, Reducer, Rolling, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    let cpu = TimeSeries::new(
        "cpu",
        (0..10).map(|i| i * 60_000).collect(),
        vec![Column::required(
            "cpu",
            vec![0.3, 0.5, 0.4, 0.9, 0.6, 0.2, 0.7, 0.8, 0.5, 0.4],
        )],
    )?;

    let centered = cpu.rolling_with(
        "5m",
        &[Agg::of("cpu", Reducer::Avg)],
        Rolling {
            alignment: Alignment::Centered,
            ..Rolling::default()
        },
    )?;
    // At minute 4 the window [1.5m, 6.5m) holds minutes 2 to 6.
    assert_eq!(
        centered.column("cpu")?.numbers()?.get(4),
        Some(0.5599999999999999)
    );
    Ok(())
}

rolling_with takes the options as a Rolling struct; Rolling::default() gives what rolling does.

Near the series' ends the window holds fewer events (the first ones under trailing, both ends under centered, the last ones under leading), but every row still gets a value from the events it does hold. To leave those rows undefined until the window is full enough, set minSamples (below).

Warm-up gate (minSamples)

By default rolling emits a result for every source event, even when the window holds only one or two samples. For stats whose stability depends on having enough data — rolling stdev, percentiles, the avg/sd pair behind a band chart — those early rows are noisy and can false-flag downstream consumers (anomaly bands collapsing around a tiny early sample). minSamples suppresses output until the window contains at least N source events:

TypeScript

// Don't trust the rolling stats until at least 20 samples are in.
cpu.rolling(
  '1m',
  { mean: { from: 'cpu', using: 'avg' }, sd: { from: 'cpu', using: 'stdev' } },
  { minSamples: 20 },
);

Rust

use pond_rs::{Agg, Column, Reducer, Rolling, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    // One reading a second for 30 seconds.
    let cpu = TimeSeries::new(
        "cpu",
        (0..30).map(|i| i * 1_000).collect(),
        vec![Column::required(
            "cpu",
            (0..30)
                .map(|i| 50.0 + ((i * 7) % 11) as f64)
                .collect::<Vec<_>>(),
        )],
    )?;

    // Don't trust the rolling stats until at least 20 samples are in.
    let stats = cpu.rolling_with(
        "1m",
        &[
            Agg::new("mean", "cpu", Reducer::Avg),
            Agg::new("sd", "cpu", Reducer::Stdev),
        ],
        Rolling {
            min_samples: 20,
            ..Rolling::default()
        },
    )?;
    let mean = stats.column("mean")?.numbers()?;
    let sd = stats.column("sd")?.numbers()?;
    // Rows 0 to 18 hold fewer than 20 samples: gaps in every column.
    assert_eq!(mean.get(18), None);
    assert_eq!(sd.get(18), None);
    // Row 19 is the first with 20.
    assert_eq!(mean.get(19), Some(54.9));
    assert_eq!(sd.get(19), Some(3.2388269481403293));
    assert_eq!(stats.len(), 30); // the length doesn't change
    Ok(())
}

Rows below the threshold emit undefined for every reducer column; the output schema and length don't change. Defaults to 0 (no gate). Same option is available on baseline() and outliers() and threads through to their internal rolling pass.

Keeping the source columns (append)

The per-event form returns the key and the reducer outputs only. Pass append: true to keep every column of the source and add the outputs after them, so a chain can keep reading the source columns, the way the financial studies add a column:

TypeScript

import '@pond-ts/financial/fluent'; // adds .rsi() to TimeSeries

const out = bars
  .rsi({ period: 14 })
  .rolling('10m', { rsi10m: { from: 'rsi', using: 'avg' } }, { append: true });
// out has time, close, rsi, rsi10m

Rust

use pond_rs::financial::Study;
use pond_rs::{Agg, Column, Reducer, Rolling, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    // 30 one-minute bars of made-up prices.
    let bars = TimeSeries::new(
        "bars",
        (0..30).map(|i| i * 60_000).collect(),
        vec![Column::required(
            "close",
            (0..30)
                .map(|i| 100.0 + (i % 7) as f64 - (i % 3) as f64 * 1.5)
                .collect::<Vec<_>>(),
        )],
    )?;

    let out = bars.rsi(Study::period(14))?.rolling_with(
        "10m",
        &[Agg::new("rsi10m", "rsi", Reducer::Avg)],
        Rolling {
            append: true,
            ..Rolling::default()
        },
    )?;
    // out has time, close, rsi, rsi10m
    let names: Vec<&str> = out.columns().iter().map(|c| c.name()).collect();
    assert_eq!(names, ["close", "rsi", "rsi10m"]);
    assert_eq!(
        out.column("rsi10m")?.numbers()?.get(29),
        Some(50.92088286604516)
    );

    // An output named after an existing column is an error with append.
    let clash = bars.rsi(Study::period(14))?.rolling_with(
        "10m",
        &[Agg::of("rsi", Reducer::Avg)],
        Rolling {
            append: true,
            ..Rolling::default()
        },
    );
    assert!(clash.is_err());
    Ok(())
}

In Rust the studies are methods on TimeSeries already; there is nothing to opt in to.

An output may not take an existing column's name, so the shorthand { rsi: 'avg' } (whose output is named after its source) is an error with append; name each output. That is a runtime error, not a type error. Not available on the grid form, whose rows are the sequence's points rather than the source's events.

Two output shapes

Pick the one that matches your output cadence:

TypeScript

// Per-event rolling: one output per source event, same key type.
cpu.rolling('5m', { cpu: 'avg' });

// Grid-anchored rolling: one output per Sequence point.
cpu.rolling(Sequence.every('1m'), '5m', { cpu: 'avg' });

Rust

use pond_rs::{Agg, Column, Reducer, RollingOver, Sequence, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    // Readings at uneven times.
    let cpu = TimeSeries::new(
        "cpu",
        vec![0, 50_000, 130_000, 140_000, 260_000, 330_000],
        vec![Column::required("cpu", vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0])],
    )?;

    // Per-row rolling: one output per source row, same key.
    let per_row = cpu.rolling("2m", &[Agg::of("cpu", Reducer::Avg)])?;
    assert_eq!(per_row.key().begin(), cpu.key().begin());

    // Grid rolling: one output per minute of the sequence.
    let grid = cpu.rolling_over(
        &Sequence::every("1m")?,
        "2m",
        &[Agg::of("cpu", Reducer::Avg)],
        RollingOver::default(),
    )?;
    assert_eq!(grid.key_name(), "interval");
    assert_eq!(
        grid.key().begin(),
        &[0, 60_000, 120_000, 180_000, 240_000, 300_000]
    );
    let got: Vec<_> = grid.column("cpu")?.numbers()?.iter().collect();
    assert_eq!(
        got,
        [
            Some(1.0),
            Some(1.5),
            Some(2.0),
            Some(3.5),
            Some(3.5),
            Some(5.0)
        ]
    );
    Ok(())
}

The grid form is rolling_over in Rust.

The per-event form is the common one. Use the grid form when you want a fixed output cadence independent of where source events land — common for chart sampling.

Mapping shapes

Same two shapes as aggregate:

TypeScript

// Shorthand
cpu.rolling('5m', { cpu: 'avg' });

// Explicit { from, using, kind? } — rename, multi-reducer-per-column
cpu.rolling('5m', {
  avg: { from: 'cpu', using: 'avg' },
  sd: { from: 'cpu', using: 'stdev' },
  peak: { from: 'cpu', using: 'max' },
});

Rust

use pond_rs::{Agg, Column, Reducer, TimeSeries};

fn main() -> Result<(), pond_rs::PondError> {
    let cpu = TimeSeries::new(
        "cpu",
        (0..10).map(|i| i * 60_000).collect(),
        vec![Column::required(
            "cpu",
            vec![0.3, 0.5, 0.4, 0.9, 0.6, 0.2, 0.7, 0.8, 0.5, 0.4],
        )],
    )?;

    // Shorthand: the output takes the source's name.
    let short = cpu.rolling("5m", &[Agg::of("cpu", Reducer::Avg)])?;
    assert!(short.column("cpu").is_ok());

    // Named outputs: several reducers on one column, one pass.
    let band = cpu.rolling(
        "5m",
        &[
            Agg::new("avg", "cpu", Reducer::Avg),
            Agg::new("sd", "cpu", Reducer::Stdev),
            Agg::new("peak", "cpu", Reducer::Max),
        ],
    )?;
    let names: Vec<&str> = band.columns().iter().map(|c| c.name()).collect();
    assert_eq!(names, ["avg", "sd", "peak"]);
    assert_eq!(band.column("avg")?.numbers()?.get(9), Some(0.52));
    assert_eq!(
        band.column("sd")?.numbers()?.get(9),
        Some(0.21354156504062624)
    );
    assert_eq!(band.column("peak")?.numbers()?.get(9), Some(0.8));
    Ok(())
}

Agg::new(output, source, reducer) is the explicit form; Agg::of(source, reducer) is the shorthand.

The explicit form is how you get a band chart in one pass — avg and sd from the same source column without two rolling traversals. Anomaly detection ships a baseline helper that's sugar over this exact pattern.

Custom reducers

Same contract as aggregate:

cpu.rolling('5m', {
  cpu: (values) => {
    const nums = values.filter((v): v is number => typeof v === 'number');
    return nums.length === 0
      ? undefined
      : nums.reduce((a, b) => a + b, 0) / nums.length;
  },
});

Mix built-in and custom reducers in the same mapping.

pond-rs has no custom reducer functions yet; it takes the built-in reducers only.

Complexity

Most built-in reducers are O(1) per event via incremental add/remove. min / max are O(1) amortized. Median and percentiles are O(N) per event (sorted-array maintenance). Full table on Reducer reference → Rolling window behavior.

tail(duration) — temporal slice

Often paired with rolling for a "current state" readout:

TypeScript

// Trailing 30s, then collapse to scalars.
const recent = series.tail('30s').reduce({
  cpu: 'p95',
  host: 'unique',
});

Rust

use pond_rs::{Agg, Column, Reducer, Scalar, TimeSeries, Value};

fn main() -> Result<(), pond_rs::PondError> {
    // A reading every 5 seconds from three hosts in turn.
    let series = TimeSeries::new(
        "cpu",
        (0..12).map(|i| i * 5_000).collect(),
        vec![
            Column::required(
                "cpu",
                vec![
                    0.2, 0.4, 0.9, 0.3, 0.5, 0.6, 0.1, 0.8, 0.7, 0.35, 0.45, 0.95,
                ],
            ),
            Column::required(
                "host",
                (0..12).map(|i| ["a", "b", "c"][i % 3]).collect::<Vec<_>>(),
            ),
        ],
    )?;

    // Trailing 30s, then collapse to single values.
    let recent = series.tail("30s")?.reduce(&[
        Agg::new("cpu", "cpu", Reducer::Percentile(95.0)),
        Agg::new("host", "host", Reducer::Unique),
    ])?;
    assert_eq!(recent.number("cpu")?, Some(0.9125));
    let hosts = vec![Scalar::from("a"), Scalar::from("b"), Scalar::from("c")];
    assert_eq!(recent.get("host")?, Some(Value::Array(hosts)));
    Ok(())
}

reduce returns a Reduced, read by output name. Unique gives a list, so it comes back as a Value::Array.

tail(duration) is the temporal counterpart to Array.slice(-n): it keeps events whose begin() is strictly greater than lastEvent.begin() - duration. Called with no argument it's the identity (whole series). Composes with every method, not just reduce — plot the last 30s, aggregate just the last hour, etc.

Live rolling aggregation

The streaming counterpart. Maintains a sliding window incrementally; each source event produces an output event with the rolling aggregate at that point.

Live: points stream into a LiveSeries (blue), the trailing window (orange) rides the newest point, and the rolling average traces the teal line — updating as data arrives. Change the window to smooth more or less. Live rolling is trailing-only: with no future events to average, the window can only look back, so the anchor always sits at its leading edge. (See @pond-ts/charts for how the chart is built.)

TypeScript

import { LiveSeries } from 'pond-ts';

const live = new LiveSeries({ name: 'cpu', schema });

const rolling = live.rolling('5m', { cpu: 'avg' });
// rolling is a LiveRollingAggregation; satisfies LiveSource

rolling.on('event', (event) => {
  console.log('rolling avg:', event.get('cpu'));
});

rolling.value(); // Record<string, ColumnValue | undefined>, e.g. { cpu: 0.42 }
//   (each value is `undefined` until the window reaches minSamples)

Rust

use pond_rs::live::{LiveSeries, Output, OutputKind};
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("cpu", ColumnKind::Number),
        ],
    )?;

    let rolling = live.rolling(live.root(), "5m", &[Agg::of("cpu", Reducer::Avg)])?;
    // Two stats from one column in one rolling pass.
    let band = live.rolling(
        live.root(),
        "1m",
        &[
            Agg::new("mean", "cpu", Reducer::Avg),
            Agg::new("sd", "cpu", Reducer::Stdev),
        ],
    )?;
    // pond-rs has no listeners: watch the node, and each push returns
    // what it emitted.
    live.watch(rolling, OutputKind::Event)?;

    let mut seen = Vec::new();
    for (i, v) in [0.2, 0.4, 0.9, 0.3].into_iter().enumerate() {
        for out in live.push(i as i64 * 30_000, [Some(v.into())])? {
            if let Output::Event { event, .. } = out {
                seen.push(event.number("cpu")?);
            }
        }
    }
    assert_eq!(
        seen,
        [Some(0.2), Some(0.30000000000000004), Some(0.5), Some(0.45)]
    );

    // The current values, one per column, in order.
    assert_eq!(live[rolling].value(), [Some(0.45.into())]);
    assert_eq!(
        live[band].value(),
        [
            Some(0.5333333333333333.into()),
            Some(0.26246692913372704.into())
        ]
    );
    Ok(())
}

pond-rs calls none of your code during a push. Instead of on('event'), watch the node and read the outputs each push returns (see Live series).

Both mapping shapes work on live the same way they work on batch — the { from, using } spec form is how you get multiple stats from one source column in a single rolling pass:

const band = live.rolling('1m', {
  mean: { from: 'cpu', using: 'avg' },
  sd: { from: 'cpu', using: 'stdev' },
});
band.value(); // { mean, sd } in one pass — no second rolling deque

The same shape works on live.aggregate(seq, ...) and on the synchronised partitioned form (live.partitionBy(col).rolling(window, ..., { trigger: Trigger.clock(seq) })).

Custom-function reducers on live: O(N) per snapshot

Custom (values) => ... reducers work on live rolling and live aggregation (since v0.14.1) but pay an O(window-size) cost at each snapshot() — the function re-runs over every value in the current window each time the accumulator emits. For high-throughput streams prefer a built-in ('avg', 'p95', 'samples', …) or compose multiple built-ins via { from, using } aliases. See the Custom reducers entry on the reducer reference for the full trade-off.

The output buffer grows unbounded — compose with window() to bound it.

Heads up on late events

Under ordering: 'reorder', a late event becomes a new output event at its insertion point — LiveRollingAggregation does not re-scan historical windows to include it. See Live transforms → Late-event scope for the full picture.

Reporting at a regular cadence: clock triggers

The live-side rolling() doesn't accept a Sequence directly the way batch rolling(seq, window, mapping) does — emission shapes differ (per-event by default for live, per-bucket for batch). When you want to report the rolling state at a regular cadence (e.g. push p95 to a backend every 30 s), set a Trigger.every(duration):

TypeScript

const rolling = live.rolling(
  '1m',
  { latency: 'p95' },
  { trigger: Trigger.every('30s') },
);

Rust

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

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

    let rolling = live.rolling_with(
        live.root(),
        "1m",
        &[Agg::new("latency", "latency", Reducer::Percentile(95.0))],
        &RollingOptions::new().trigger(Trigger::every("30s")?),
    )?;

    for (t, ms) in [
        (0, 100.0),
        (10_000, 120.0),
        (25_000, 300.0),
        (31_000, 110.0),
        (45_000, 90.0),
        (62_000, 130.0),
    ] {
        live.push(t, [Some(ms.into())])?;
    }

    // One row each time an event crosses into a new 30 s, keyed at its start.
    let rows: Vec<_> = live[rolling]
        .events()
        .map(|e| (e.begin(), e.number("latency").unwrap()))
        .collect();
    assert_eq!(
        rows,
        [(30_000, Some(272.99999999999994)), (60_000, Some(266.0))]
    );
    Ok(())
}

For the full trigger reference (event / clock / count, semantics, sync partitioned rolling), see Triggering. The Telemetry reporting recipe is the end-to-end worked example.

Multi-window rolling

When a use case needs several trailing windows over the same source — a 1-minute baseline of avg/stdev plus a 200-ms current-tick window of raw samples for anomaly detection, say — pass a record-of-mappings instead of a single (window, mapping) pair:

TypeScript

const fused = live.rolling(
  {
    '1m': {
      cpu_avg: { from: 'cpu', using: 'avg' },
      cpu_sd: { from: 'cpu', using: 'stdev' },
    },
    '200ms': { cpu_samples: { from: 'cpu', using: 'samples' } },
  },
  { trigger: Trigger.every('200ms') },
);

fused.on('event', (e) => {
  // All four columns from both windows on a single event.
  const samples = e.get('cpu_samples') as ReadonlyArray<number>;
  const z = samples.map((s) => (s - e.get('cpu_avg')) / e.get('cpu_sd'));
  // …
});

Rust

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

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

    let fused = live.rolling_windows(
        live.root(),
        &[
            RollingWindow::new(
                "1m",
                &[
                    Agg::new("cpu_avg", "cpu", Reducer::Avg),
                    Agg::new("cpu_sd", "cpu", Reducer::Stdev),
                ],
            ),
            RollingWindow::new("200ms", &[Agg::new("cpu_samples", "cpu", Reducer::Samples)]),
        ],
        &RollingOptions::new().trigger(Trigger::every("200ms")?),
    )?;

    for (t, v) in [
        (0, 1.0),
        (100, 2.0),
        (150, 3.0),
        (250, 5.0),
        (320, 4.0),
        (410, 6.0),
    ] {
        live.push(t, [Some(v.into())])?;
    }

    // All three columns from both windows on one event.
    let e = live[fused].at(-1).unwrap();
    assert_eq!(e.begin(), 400);
    let (avg, sd) = (e.number("cpu_avg")?.unwrap(), e.number("cpu_sd")?.unwrap());
    assert_eq!((avg, sd), (3.5, 1.707825127659933));
    let samples = e.array("cpu_samples")?.unwrap();
    assert_eq!(samples, [Scalar::from(5.0), 4.0.into(), 6.0.into()]);
    let z: Vec<f64> = samples
        .iter()
        .map(|s| match s {
            Scalar::Number(s) => (s - avg) / sd,
            _ => f64::NAN,
        })
        .collect();
    assert_eq!(z.len(), 3);
    Ok(())
}

The keyed form is rolling_windows in Rust, one RollingWindow per window.

This is one rolling primitive that maintains all declared windows in one ingest pass over a single shared deque, then emits one merged event per trigger fire with all windows' columns concatenated into one record.

The naive way before this primitive existed — two separate rolling() calls plus a per-(ts, key) join — works but pays every per-event pipeline cost twice (and N times for N windows). The keyed form does the per-event work once.

Properties

• One merged output stream. All columns from all windows arrive on a single event. Consumer code is one event handler, not a multi-stream join.
• Single trigger across all windows. Per-window cadence is not supported — that's what the fusion saves. Users who need different cadences per window fall back to two separate rolling() calls.
• Time-based windows only. Object keys are duration strings ('1m', '200ms', '5s'); count-based windows (live.rolling(100, ...)) stay on the single-window form and aren't mixable here.
• Duplicate output column names across windows are rejected at construction.
• Single-window equivalence. live.rolling('1m', m, opts) and live.rolling({ '1m': m }, opts) are semantically identical — pinned by tests so the unification is incomplete the moment it diverges.

Per-window options

When one window needs different options from the rest (minSamples, e.g.), the value form switches from a bare mapping to an elaborated wrapper:

TypeScript

live.rolling(
  {
    '1m': { cpu_avg: { from: 'cpu', using: 'avg' } },
    '200ms': {
      mapping: { cpu_samples: { from: 'cpu', using: 'samples' } },
      minSamples: 5, // per-window override
    },
  },
  { trigger, minSamples: 2 }, // top-level default
);

Rust

use pond_rs::live::{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("cpu", ColumnKind::Number),
        ],
    )?;

    let fused = live.rolling_windows(
        live.root(),
        &[
            RollingWindow::new("1m", &[Agg::new("cpu_avg", "cpu", Reducer::Avg)]),
            RollingWindow::new("200ms", &[Agg::new("cpu_samples", "cpu", Reducer::Samples)])
                .min_samples(5), // this window's own setting
        ],
        &RollingOptions::new()
            .trigger(Trigger::every("200ms")?)
            .min_samples(2), // the default for the rest
    )?;

    for (t, v) in [
        (0, 1.0),
        (100, 2.0),
        (150, 3.0),
        (250, 5.0),
        (320, 4.0),
        (410, 6.0),
    ] {
        live.push(t, [Some(v.into())])?;
    }

    let e = live[fused].at(-1).unwrap();
    assert_eq!(e.number("cpu_avg")?, Some(3.5));
    // The 200 ms window holds 3 events, short of its 5: a gap.
    assert_eq!(e.array("cpu_samples")?, None);
    Ok(())
}

Top-level options apply as the default across all windows; per-window elaborated minSamples overrides for that window.

Partitioned variant

The same form composes with partitionBy(...) and works through chained pipelines:

TypeScript

const fused = live
  .partitionBy('host')
  .fill({ cpu: 'hold' })
  .rolling(
    {
      '1m': { cpu_avg: { from: 'cpu', using: 'avg' } },
      '200ms': { cpu_samples: { from: 'cpu', using: 'samples' } },
    },
    { trigger: Trigger.every('200ms') },
  );
// One merged event per partition per boundary; partition column
// auto-injected at the front of the schema.

Rust

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

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

    let hosts = live.partition_by(live.root(), "host", &PartitionOptions::new())?;
    let filled = live.per_partition(hosts, |live, host| {
        Ok(live.fill(host, &[("cpu", Fill::Hold)])?.id())
    })?;
    let fused = live.partition_rolling_windows(
        filled,
        &[
            RollingWindow::new("1m", &[Agg::new("cpu_avg", "cpu", Reducer::Avg)]),
            RollingWindow::new("200ms", &[Agg::new("cpu_samples", "cpu", Reducer::Samples)]),
        ],
        &RollingOptions::new().trigger(Trigger::every("200ms")?),
    )?;

    live.push(0, [Some(1.0.into()), Some("a".into())])?;
    live.push(50, [Some(10.0.into()), Some("b".into())])?;
    live.push(100, [None, Some("a".into())])?; // held at 1
    live.push(150, [Some(20.0.into()), Some("b".into())])?;
    live.push(250, [Some(3.0.into()), Some("a".into())])?;

    // One row per host at each 200 ms boundary.
    let rows: Vec<_> = live[fused]
        .events()
        .map(|e| {
            let host = e.string("host").unwrap().unwrap().to_string();
            let samples = e.array("cpu_samples").unwrap().unwrap().to_vec();
            (e.begin(), host, e.number("cpu_avg").unwrap(), samples)
        })
        .collect();
    assert_eq!(
        rows,
        [
            (
                200,
                "a".to_string(),
                Some(1.6666666666666667),
                vec![Scalar::from(1.0), 3.0.into()]
            ),
            (
                200,
                "b".to_string(),
                Some(15.0),
                vec![Scalar::from(10.0), 20.0.into()]
            ),
        ]
    );
    Ok(())
}

In Rust, per_partition builds the fill on every host, and partition_rolling_windows runs the windows over it. As in pond-ts, it needs a clock trigger.

On the partitioned form, clock trigger is required — synced cross-partition emission needs a single shared boundary detector.

Performance

The architectural argument: per-event hops run once vs N times in N separate rollings. Bench numbers from packages/core/scripts/perf-fused-rolling.mjs:

Windows	Separate (ms)	Multi-window (ms)	Wall delta	Heap delta
2	152.91	102.91	−33%	−33%
3	186.63	79.89	−57%	−54%
4	245.42	107.51	−56%	−58%
5	279.79	118.90	−58%	−66%

100k events × 100 hosts at each N. Multi-window stays roughly constant; separate scales linearly. At N=5, multi-window is 2.4× faster and uses 34% of the heap. The win compounds — fused pipeline overhead is O(1) in the number of windows; per-window reducer state remains O(N) (which separate also pays).

When to use multi-window vs separate rollings

• Multi-window when several windows naturally emit together — baseline + current tick, day + hour summaries, etc. The shared ingest pass is the win, and the merged output is the ergonomic win.
• Separate rollings when the windows have different cadences (one fires every 30s, the other every 200ms), or when you only need one of the windows in a given consumer (you're not paying for the unused one's reducer state).

See Windowing → Multi-window rolling (live) for the conceptual framing.