Incoming events invalidate the node; they do not snapshot. The
toTimeSeries() call happens when something pulls, so a burst of
events costs one dirty mark each (O(1) after the first, since dirty
propagation cuts off at already-dirty nodes) and exactly one snapshot
at the next pull — not one snapshot per event.
That keeps the graph on the right side of pond's split: incremental
per-event computation stays in the live layer, and the graph composes
whole-value batch transforms over snapshots.
constfeed=fromLive(liveSeries); consthourly=derive({ s: feed.out.value }, ({ s }) => s.aggregate(Sequence.every('1h'), { cpu: 'avg' }), ); // ... events arrive ... hourly.out.value.get(); // one snapshot, one aggregate feed.dispose();
Bind the aggregation, not the buffer
The graph has no partial invalidation: a dirty node recomputes from a
whole snapshot, so the pipeline above re-aggregates every retained
event on every pull even though only the tail moved. Push the windowed
work down into the live layer instead, and bind its output:
LiveAggregation keeps its buckets current per event, so a pull
materializes bucket count rather than event count. Measured at 200k
events through a 50k-event buffer, pulling every 1k events: 9.05 ms
per pull re-aggregating the buffer, 0.04 ms per pull off the live
aggregation — 235x. The gap widens with buffer size, because the
first is O(retained events) and the second is O(buckets).
This is a semantic change, not just a faster path. A live
aggregation exposes closed buckets. Data is the clock, so the
newest bucket stays invisible until an event crosses its end, while
re-aggregating the raw buffer includes that partial tail bucket
immediately. Two hours of minute data ending at 1h59m reads as one
row through the aggregation and two through the buffer. If the
current, still-filling bucket has to be on screen, keep
re-aggregating the buffer and pay for it — or drive emission with a
Trigger so buckets close on a schedule you control.
This is why fromLive takes a GraphSource rather than a
SnapshotSource: the incremental operators are precisely the
ones without a toTimeSeries() method, and excluding them would rule
out the only answer to the cost above.
Binds a pond live source into a graph.
Incoming events invalidate the node; they do not snapshot. The
toTimeSeries()call happens when something pulls, so a burst of events costs one dirty mark each (O(1) after the first, since dirty propagation cuts off at already-dirty nodes) and exactly one snapshot at the next pull — not one snapshot per event.That keeps the graph on the right side of pond's split: incremental per-event computation stays in the live layer, and the graph composes whole-value batch transforms over snapshots.
Bind the aggregation, not the buffer
The graph has no partial invalidation: a dirty node recomputes from a whole snapshot, so the pipeline above re-aggregates every retained event on every pull even though only the tail moved. Push the windowed work down into the live layer instead, and bind its output:
LiveAggregationkeeps its buckets current per event, so a pull materializes bucket count rather than event count. Measured at 200k events through a 50k-event buffer, pulling every 1k events: 9.05 ms per pull re-aggregating the buffer, 0.04 ms per pull off the live aggregation — 235x. The gap widens with buffer size, because the first is O(retained events) and the second is O(buckets).This is a semantic change, not just a faster path. A live aggregation exposes closed buckets. Data is the clock, so the newest bucket stays invisible until an event crosses its end, while re-aggregating the raw buffer includes that partial tail bucket immediately. Two hours of minute data ending at 1h59m reads as one row through the aggregation and two through the buffer. If the current, still-filling bucket has to be on screen, keep re-aggregating the buffer and pay for it — or drive emission with a
Triggerso buckets close on a schedule you control.This is why
fromLivetakes a GraphSource rather than a SnapshotSource: the incremental operators are precisely the ones without atoTimeSeries()method, and excluding them would rule out the only answer to the cost above.