Windows & Aggregations Reference

Complete reference for window types and aggregation functions in VPL.

When windows close

Time windows are judged in the time the events carry. A program keeps a clock per input event type, the latest timestamp read of that type, and a window closes once the clocks of the types feeding it (through any chain of streams) have passed its end, whichever stream or partition moved them. A brute force counted per address on Security events raises its alert on the next Security event of any kind, not when the same address fails again, which after a successful logon may never happen.

• A window fed by several types waits for the slowest of them, so a source that delivers late does not see its windows closed by a faster one. A type that never arrived does not hold anything back, and a quiet type only holds back the windows it feeds.
• Windows close after each batch of input (for varpulis simulate and a Vejas detect unit, after each event), as an event of their own arriving at that time would close them. A sliding window emits at each slide reached.
• What a window emits reaches the streams below it before their own windows close, so a count over a count, or a correlation over a count, sees it in the right window.
• .watermark(out_of_order: 30s) on a stream holds its windows 30 seconds longer, for events that arrive late.
• At the end of a bounded input (varpulis simulate), every window closes.

A live host can also let time pass when a source goes quiet: with an idle grace set (Program::set_idle_grace; a Vejas detect unit sets VEJAS_IDLE_CLOSE_SECS, 60 seconds by default), a type that has sent nothing for the grace has its event time move on with the wall clock, less the grace, and the host calls Program::tick() while nothing arrives. A brute force on a sparse source is then raised about a grace after its window ends. The type's clock is kept in the program's snapshot, so a restarted host picks up where it was. varpulis simulate sets no grace: a replay is judged in event time alone, and gives the same alerts however fast it runs.

An absence (-> NOT B within 4h, see Absence) completes on the same clocks, once they pass its deadline, and the idle grace raises it on a quiet source too. The end of the input completes none.

This is the synchronous engine, the one varpulis simulate and a Vejas detect unit run. The legacy asynchronous runtime still closes a window when one of its own events arrives, or on .watermark().

Window Types

Tumbling Windows

Non-overlapping, fixed-duration windows. When the duration has passed in event time, the window emits and resets.

Syntax:

stream Name = EventType
    .window(<duration>)
    .aggregate(...)

Parameters:

• <duration>: Window length (e.g., 5s, 1m, 1h)

Behavior:

• The window opens with its first event and lasts <duration> of event time
• It closes once event time reaches window_start + duration: when any later event of the types feeding it arrives, from any stream or partition, not only one of its own (see When windows close)
• The next event of the stream opens the next window

Example:

stream MinuteStats = SensorReading
    .window(1m)
    .aggregate(
        avg_value: avg(value),
        max_value: max(value),
        min_value: min(value),
        count: count()
    )

Implementation: TumblingWindow struct in varpulis-runtime/src/window.rs:15

---

Sliding Windows

Overlapping windows that slide at a specified interval, providing a rolling view of data.

Syntax:

stream Name = EventType
    .window(<size>, sliding: <slide>)
    .aggregate(...)

Parameters:

• <size>: Window size (e.g., 5m)
• <slide>: Slide interval (e.g., 1m)

Behavior:

• Maintains events within the window size
• Emits current window contents at each slide interval
• Old events are evicted as they fall outside the window
• Window contents overlap between emissions

Example:

stream RollingAverage = TemperatureReading
    .window(5m, sliding: 30s)
    .aggregate(
        rolling_avg: avg(temperature),
        recent_max: max(temperature)
    )

Implementation: SlidingWindow struct in varpulis-runtime/src/window.rs:77

---

Count Windows

Windows based on event count rather than time.

Syntax:

stream Name = EventType
    .window(<n>)
    .aggregate(...)

Parameters:

• <n>: Number of events per window

Behavior:

• Collects exactly N events
• Emits when count is reached
• Resets to empty after emission

Example:

stream BatchProcessor = Transaction
    .window(100)
    .aggregate(
        batch_total: sum(amount),
        batch_count: count()
    )

Implementation: CountWindow struct in varpulis-runtime/src/window.rs:146

---

Sliding Count Windows

Count-based windows with overlap.

Syntax:

stream Name = EventType
    .window(<size>, sliding: <slide>)
    .aggregate(...)

Parameters:

• <size>: Window size in events
• <slide>: Slide size in events

Example:

stream RollingBatch = Reading
    .window(100, sliding: 25)
    .aggregate(
        rolling_sum: sum(value)
    )

Implementation: SlidingCountWindow struct in varpulis-runtime/src/window.rs:194

---

Partitioned Windows

Any window type can be partitioned by a key field.

Syntax:

stream Name = EventType
    .partition_by(<field>)
    .window(<params>)
    .aggregate(...)

In a pattern, the key may be written with the step alias that binds it — .partition_by(login.user_id) — which means the same thing as .partition_by(user_id): partition keys are resolved per incoming event, before any alias is bound, so the alias is documentation for the reader. The alias must name a step of that pattern.

A computed key (.partition_by(a + b), .partition_by(lower(host))) is refused at compile time. It used to be dropped on the floor, leaving a global window and an unpartitioned pattern engine. Derive the value into a field with .select() or .emit() first, then partition on that field.

Behavior:

• Creates independent windows per unique partition key
• Each partition emits separately
• Memory scales with number of unique keys

Example:

stream PerDeviceStats = SensorReading
    .partition_by(device_id)
    .window(1m)
    .aggregate(
        device: last(device_id),
        avg_reading: avg(value),
        readings: count()
    )

Implementations:

• PartitionedTumblingWindow in window.rs
• PartitionedSlidingWindow in window.rs

---

Duration Units

Unit	Abbreviation	Example
Milliseconds	ms	500ms
Seconds	s	30s
Minutes	m	5m
Hours	h	1h
Days	d	1d

Examples:

.window(500ms)  # Half second
.window(30s)  # 30 seconds
.window(5m, sliding: 1m)  # 5 min window, 1 min slide
.window(1h)  # 1 hour
.window(1d)  # 1 day

---

Aggregation Functions

All aggregations operate on events within a window.

count()

Count the number of events in the window.

Signature: count() -> int

Example:

.aggregate(total_events: count())

---

sum(field)

Sum numeric field values. SIMD-optimized on x86_64 with AVX2.

Signature: sum(field | expression) -> float

The argument may be a field name or an expression evaluated per event, so sum(price * quantity) sums the products rather than requiring a precomputed column. The same holds for avg, min, max, stddev, the percentiles, and count_distinct.

Example:

.aggregate(
    total_amount: sum(amount),
    total_quantity: sum(quantity),
    notional: sum(price * quantity)
)

Performance: Uses SIMD (AVX2) for 4x parallel f64 addition when available.

---

avg(field)

Compute the arithmetic mean. SIMD-optimized.

Signature: avg(field: string) -> float | null

Returns: null if no events contain the field.

Example:

.aggregate(average_price: avg(price))

Performance: Extracts values to contiguous array for SIMD summation.

---

min(field)

Find the minimum value. SIMD-optimized.

Signature: min(field: string) -> float | null

Example:

.aggregate(lowest_temp: min(temperature))

Performance: Uses SIMD _mm256_min_pd for parallel comparison.

---

max(field)

Find the maximum value. SIMD-optimized.

Signature: max(field: string) -> float | null

Example:

.aggregate(highest_temp: max(temperature))

Performance: Uses SIMD _mm256_max_pd for parallel comparison.

---

stddev(field)

Compute the sample standard deviation using Welford's online algorithm.

Signature: stddev(field: string) -> float | null

Returns: null if fewer than 2 values.

Example:

.aggregate(temp_variance: stddev(temperature))

Algorithm: Single-pass Welford's algorithm for numerical stability.

---

percentile(field, q)

Compute the q-th percentile using linear interpolation. The quantile q is specified as a float between 0.0 and 1.0.

Signature: percentile(field: string, q: float) -> float | null

Example:

.aggregate(
    p50_latency: percentile(latency, 0.5),
    p95_latency: percentile(latency, 0.95),
    p99_latency: percentile(latency, 0.99),
    p999_latency: percentile(latency, 0.999)
)

Algorithm: Sort-based with linear interpolation. Collects all f64 values, sorts, then interpolates at the quantile position.

---

median(field)

Shorthand for the 50th percentile.

Signature: median(field: string) -> float | null

Example:

.aggregate(median_price: median(price))

---

p50(field) / p95(field) / p99(field)

Convenience aliases for common percentiles.

Signature: p50(field) -> float | null (equivalent to percentile(field, 0.5))

Example:

.aggregate(
    p50_latency: p50(latency),
    p95_latency: p95(latency),
    p99_latency: p99(latency)
)

---

collect(field)

Collect all field values into an array.

Signature: collect(field: string) -> array

Example:

.aggregate(
    all_sensors: collect(sensor_id),
    all_values: collect(value)
)

---

first(field)

Return the first value in the window (by event order).

Signature: first(field: string) -> value | null

Example:

.aggregate(
    window_start: first(timestamp),
    initial_value: first(value)
)

---

last(field)

Return the last value in the window (by event order).

Signature: last(field: string) -> value | null

Example:

.aggregate(
    window_end: last(timestamp),
    final_value: last(value)
)

---

distinct(field) / count_distinct(field)

Count unique values of a field.

Signature: count(distinct(field: string)) -> int

Example:

.aggregate(
    unique_users: count(distinct(user_id)),
    unique_products: count(distinct(product_id))
)

---

Incremental Aggregations

For high-throughput scenarios, Varpulis supports incremental aggregation where possible.

IncrementalSum

O(1) updates for sum instead of O(n) recomputation.

Supported operations:

• add(value) - Add a value: O(1)
• remove(value) - Remove a value: O(1)
• sum() - Get current sum: O(1)
• count() - Get current count: O(1)
• avg() - Get current average: O(1)

Use case: Sliding windows where events expire frequently.

Implementation: IncrementalSum in varpulis-runtime/src/simd.rs:391

---

IncrementalMinMax

Maintains min/max with incremental updates.

Supported operations:

• add(value) - Add a value: O(1) amortized
• remove(value) - Remove a value: O(n) worst case
• min() - Get current min: O(1) after sort
• max() - Get current max: O(1) after sort

Implementation: IncrementalMinMax in varpulis-runtime/src/simd.rs:444

---

SIMD Optimization

Aggregations use SIMD (Single Instruction Multiple Data) vectorization when available.

Requirements

• x86_64 architecture
• AVX2 instruction set support (most CPUs since 2013)

Performance Characteristics

Operation	Scalar	SIMD (AVX2)	Speedup
sum	O(n)	O(n/4)	~4x
min	O(n)	O(n/4)	~4x
max	O(n)	O(n/4)	~4x
avg	O(n)	O(n/4)	~4x

How It Works

1. Field Extraction: Numeric field values are extracted to contiguous Vec<f64>
2. Vectorization: AVX2 processes 4 f64 values per instruction
3. Loop Unrolling: 4-way unrolling for better instruction-level parallelism
4. Horizontal Reduction: Final aggregation of vector lanes
5. Remainder Handling: Scalar processing for remaining elements

Fallback

On non-AVX2 systems, scalar 4-way loop unrolling provides ~2x speedup over naive loops.

---

Memory Considerations

Window Memory Usage

Window Type	Memory Formula
Tumbling	O(events_per_window)
Sliding	O(events_per_window)
Count	O(window_size)
Partitioned	O(partitions × events_per_window)

Recommendations

1. Use appropriate window sizes: Smaller windows = less memory
2. Partition carefully: Each partition key creates a separate window
3. Consider count windows: Fixed memory usage vs. time windows
4. Monitor partition growth: Unbounded partition keys can exhaust memory

---

Examples

Real-time Monitoring

stream SensorAlerts = SensorReading
    .partition_by(sensor_id)
    .window(1m)
    .aggregate(
        sensor: last(sensor_id),
        avg_value: avg(value),
        max_value: max(value),
        readings: count()
    )
    .where(avg_value > threshold or max_value > critical_threshold)
    .emit(sensor: sensor, avg_value: avg_value, max_value: max_value)

Rolling Statistics

stream RollingStats = MarketData
    .window(5m, sliding: 10s)
    .aggregate(
        vwap: sum(price * volume) / sum(volume),
        high: max(price),
        low: min(price),
        volatility: stddev(price)
    )

Batch Processing

stream BatchReport = Transaction
    .window(1000)
    .aggregate(
        batch_num: count(),
        total_value: sum(amount),
        avg_value: avg(amount),
        unique_customers: count(distinct(customer_id))
    )
    .print("Processed batch:", batch_num, "txns,", total_value, "total")

Latency Percentiles

stream ApiLatency = RequestEvent
    .partition_by(endpoint)
    .window(1m)
    .aggregate(
        endpoint: last(endpoint),
        median: median(latency_ms),
        p95: p95(latency_ms),
        p99: p99(latency_ms),
        p999: percentile(latency_ms, 0.999),
        count: count()
    )
    .where(p99 > 500)
    .emit(
        alert: "high_latency",
        endpoint: endpoint,
        p99_ms: p99,
        request_count: count
    )

Multi-Level Aggregation

# First level: per-device minute stats
stream DeviceMinutes = SensorReading
    .partition_by(device_id)
    .window(1m)
    .aggregate(
        device: last(device_id),
        minute_avg: avg(value)
    )

# Second level: all-device hour stats
stream HourlyOverview = DeviceMinutes
    .window(1h)
    .aggregate(
        active_devices: count(distinct(device)),
        overall_avg: avg(minute_avg),
        hottest_device: max(minute_avg)
    )

---

See Also

• Language Tutorial - Learn VPL basics
• SASE+ Pattern Guide - Pattern matching over windows