Documentation
¶
Overview ¶
Package windowed expresses time-bucketed aggregations on top of any structural monoid. State is keyed by (entity, bucket_id) where bucket_id is computed from event time and the configured Granularity. Queries assemble sliding windows by Combining the N most recent buckets through the wrapped monoid. DDB TTL handles eviction past Retention.
Index ¶
Constants ¶
const DefaultMaxBucketSpan = 100_000
DefaultMaxBucketSpan caps the fan-out of a Config that sets a Granularity but neither MaxBuckets nor Retention. That shape used to get no cap at all, so a hand-built Config{Granularity: time.Minute} still let GetRange(epoch, now) build ~30 million keys before the first store call.
It is a backstop against unbounded, not a tuning knob. 100,000 buckets sits far above any deliberate query — seven days at Minute granularity is 10,080 buckets, a year at Hourly is 8,760 — while holding the key slice to a couple of megabytes instead of hundreds. A pipeline that genuinely wants to read more sets MaxBuckets and says so.
Variables ¶
This section is empty.
Functions ¶
func MergeBuckets ¶
MergeBuckets folds bucket values via the inner monoid in stable order. Used by the query layer to produce sliding-window results from per-bucket reads.
Types ¶
type Config ¶
type Config struct {
// Granularity is the size of each tumbling bucket (e.g. 24h for daily, 1h for hourly).
// Smaller granularity means finer query precision and higher state cost.
Granularity time.Duration
// Retention is how long buckets are kept before TTL eviction. Sliding-window queries
// can ask for any range up to Retention.
Retention time.Duration
// MaxBuckets caps how many buckets a single read may touch, counting both ends of
// an inclusive range. Leave it zero to take the default from Retention (see
// MaxBucketSpan).
//
// Without a cap, a caller who passes an open-ended range gets no error — just an
// enormous key list. GetRange(entity, Unix(0,0), now) against Minute granularity
// builds 29,797,201 keys (~715MB of state.Key) before the first store call, and
// all but the last Retention/Granularity of them address buckets that TTL evicted
// and can never hold data again.
MaxBuckets int
}
Config describes the bucket layout for a windowed aggregation.
The timestamp a record is bucketed by is source.Record.EventTime — every runtime passes it to BucketID, falling back to the wall clock only when a record carries no event time. There is no per-field extractor here: sources own event-time, not the window config.
func Daily ¶
Daily returns a Config with 24h granularity and the given retention. The most common configuration for "last N days of X"-style counters.
func Hourly ¶
Hourly returns a Config with 1h granularity and the given retention. Suitable for "last N hours" or "last 7 days at hourly resolution" queries.
func Minute ¶
Minute returns a Config with 1-minute granularity and the given retention. Useful for high-resolution short-window aggregations (last 5 minutes, last hour at per-minute resolution). At this granularity, "last 7 days" reads 10080 buckets per query — consider hierarchical roll-ups for queries spanning more than ~24h.
func (Config) BucketID ¶
BucketID assigns the given time to a bucket according to Granularity. Buckets are tumbling and aligned to the Unix epoch.
func (Config) BucketRange ¶
BucketRange returns the inclusive range of bucket IDs that cover [start, end].
func (Config) LastN ¶
LastN returns the bucket-ID range covering the most recent d duration ending at now. The number of buckets returned is ceil(d / Granularity); the upper bound is the bucket containing now and the range extends backward that many buckets. So for daily granularity, "last 7 days" returns 7 buckets (today plus 6 prior), not 8.
func (Config) MaxBucketSpan ¶ added in v0.2.0
MaxBucketSpan reports the largest number of buckets one read may touch, counting both ends of an inclusive range.
MaxBuckets wins when set: it is a bucket count the operator wrote down literally. Otherwise the cap is derived from Retention as ceil(Retention/Granularity) + 1, and that +1 is load-bearing. Buckets are tumbling and BucketRange is inclusive at both ends, so an absolute [t, t+Retention] touches Retention/Granularity + 1 of them. Deriving ceil(Retention/Granularity) rejected a read over exactly the retention window — the most natural range a caller writes.
With neither field set the cap is DefaultMaxBucketSpan. Zero — genuinely unbounded — comes back only for a Config with no Granularity, which assigns everything to bucket 0 and so cannot fan out at all.
func (Config) RetentionBuckets ¶ added in v0.2.0
RetentionBuckets reports how many whole buckets fit inside Retention — the longest window a read can ask for and still be answered entirely from live buckets. Zero means Retention is unset (or shorter than a single bucket), in which case callers should not enforce a retention bound.
Source Files