Barograph.Aggregate (barograph v0.1.0)

Copy Markdown View Source

Continuous aggregates: partial state, watermark refresh, invalidation (spec §8).

Aggregates store count/sum, never avg — the rule that makes hierarchical rollups compose and re-aggregation over arbitrary windows stay correct (§8.2). Refresh aggregates only (watermark, now - lag] and upserts, so the cost is proportional to new data, not total data (§8.3). Late samples below the watermark mark their bucket dirty in bg_agg_invalid and the next refresh recomputes it (§8.4).

Known v0.1 limitation: sum_dt/sum_v_dt (time-weighted average state) are computed from intervals within the refreshed window only — the interval between the last sample before a window and its first sample inside is not counted.

Summary

Types

definition()

Aggregate definition, one row of bg_agg_meta.

Functions

create(conn, name, opts, time_unit)

Creates the rollup table bg_agg_<name> and registers the aggregate in bg_agg_meta. The watermark starts at zero, so the first refresh backfills all existing data up to now - lag (Timescale behaviour).

definitions(conn)

All registered aggregate definitions.

mark_invalidations(conn, rows)

Marks dirty buckets for late-arriving samples (spec §8.4): one row in bg_agg_invalid per affected aggregate and bucket. rows are {series_id, ts} pairs from a just-committed batch.

refresh(conn, defn, now)

Refreshes one aggregate: recomputes invalidated buckets below the watermark, then finalises buckets up to now - lag. Idempotent and crash-safe — rerunning after a failure redoes the same work.

Types

definition()

@type definition() :: %{
  name: String.t(),
  source: String.t(),
  bucket_width: pos_integer(),
  watermark: integer(),
  lag: non_neg_integer(),
  refresh_every: pos_integer()
}

Aggregate definition, one row of bg_agg_meta.

bucket_width, watermark, and lag are in the database's time unit; refresh_every is in milliseconds (it is a scheduling interval, not a timestamp).

Functions

create(conn, name, opts, time_unit)

@spec create(Exqlite.Sqlite3.db(), String.t(), keyword(), atom()) ::
  :ok | {:error, term()}

Creates the rollup table bg_agg_<name> and registers the aggregate in bg_agg_meta. The watermark starts at zero, so the first refresh backfills all existing data up to now - lag (Timescale behaviour).

definitions(conn)

@spec definitions(Exqlite.Sqlite3.db()) :: [definition()]

All registered aggregate definitions.

mark_invalidations(conn, rows)

@spec mark_invalidations(Exqlite.Sqlite3.db(), [{integer(), integer()}]) :: :ok

Marks dirty buckets for late-arriving samples (spec §8.4): one row in bg_agg_invalid per affected aggregate and bucket. rows are {series_id, ts} pairs from a just-committed batch.

refresh(conn, defn, now)

@spec refresh(Exqlite.Sqlite3.db(), definition(), integer()) :: :ok

Refreshes one aggregate: recomputes invalidated buckets below the watermark, then finalises buckets up to now - lag. Idempotent and crash-safe — rerunning after a failure redoes the same work.

The aggregated range is [watermark, upper) — exclusive at the top, so a sample exactly at upper stays in its (not yet complete) bucket and is picked up by a later refresh.