Aggregates

The aggregates that survive a checkpoint, count and sum through string_agg, quantile and approx_quantile, grouped over a plain GROUP BY or an event-time window.

An aggregate folds a group of rows into one. On a stream it runs incrementally, and its running state is part of the checkpoint, so the set of aggregates ubik supports is the set whose state is durable across a restart.

Every example runs against this file.

cat > sales.ndjson <<'EOF'
{"region":"EU","merchant":"acme","amount":250}
{"region":"EU","merchant":"globex","amount":42}
{"region":"EU","merchant":"acme","amount":980}
{"region":"US","merchant":"initech","amount":110}
{"region":"US","merchant":"initech","amount":7}
{"region":"US","merchant":"hooli","amount":300}
EOF

GROUP BY

A plain GROUP BY with no window holds one running value per group and emits at the end of the source.

ubik --from file://sales.ndjson \
  "SELECT region, count(*) AS n, sum(amount) AS total,
          avg(amount) AS mean, min(amount) AS lo, max(amount) AS hi
   FROM sales GROUP BY region"

{"region":"EU","n":3,"total":1272,"mean":424.0,"lo":42,"hi":980}
{"region":"US","n":3,"total":417,"mean":139.0,"lo":7,"hi":300}

Group over a window function instead, and a group emits as its window closes; see windows.

Collecting and picking

string_agg and list (aliased array_agg) collect a group's values. arg_max and arg_min pick the value of one column at the row where another is greatest or least; first and last pick by arrival order.

ubik --from file://sales.ndjson \
  "SELECT region,
          string_agg(merchant, ',') AS names,
          array_agg(amount) AS amounts,
          arg_max(merchant, amount) AS top_merchant,
          arg_min(merchant, amount) AS low_merchant
   FROM sales GROUP BY region"

{"region":"EU","names":"acme,globex,acme","amounts":[250,42,980],"top_merchant":"acme","low_merchant":"globex"}
{"region":"US","names":"initech,initech,hooli","amounts":[110,7,300],"top_merchant":"hooli","low_merchant":"initech"}

Distinct counts and quantiles

median and quantile are exact. approx_count_distinct and approx_quantile trade exactness for bounded memory, which is what makes them safe on an unbounded stream.

ubik --from file://sales.ndjson \
  "SELECT region,
          median(amount) AS med,
          quantile(amount, 0.9) AS p90,
          approx_count_distinct(merchant) AS merchants,
          approx_quantile(amount, 0.9) AS approx_p90
   FROM sales GROUP BY region"

{"region":"EU","med":250.0,"p90":980,"merchants":2,"approx_p90":980}
{"region":"US","med":110.0,"p90":300,"merchants":2,"approx_p90":300}

approx_count_distinct is a HyperLogLog counter. approx_quantile is a DDSketch with a relative error of 0.99%, the guarantee an SLO on a latency column is written against. Both are deterministic on replay, so a resumed run reproduces the pre-crash number exactly.

The durable set

These are the aggregates that can run on a stream, because their state checkpoints:

count, sum, avg, min, max, first, last, any_value, arg_min, arg_max, string_agg, list / array_agg, exact quantile / median, approx_count_distinct, approx_quantile.

An aggregate whose state cannot be checkpointed is refused at plan time, before the first row, rather than silently breaking exactly-once at the checkpoint boundary.

ubik --from file://sales.ndjson \
  "SELECT region, reservoir_quantile(amount, 0.5) AS p FROM sales GROUP BY region"

{"error":{"code":"DURABILITY_NOT_SUPPORTED","message":"aggregate 'reservoir_quantile' cannot be checkpointed on streams: its state is not durable across a restart, which would break exactly-once. Supported: count/sum/avg, min/max, first/last/any_value, arg_min/arg_max, string_agg, list/array_agg, exact quantile/median, approx_count_distinct, and approx_quantile."}}

Order-dependent aggregates and HOP

string_agg, list, first, last, arg_min and arg_max depend on the order rows arrive. A HOP window puts each row in several overlapping windows, which cannot share one ordered accumulator, so these are refused on HOP:

NOT_SUPPORTED: aggregate 'string_agg' is order-dependent and cannot span HOP windows; use it with TUMBLE or a plain GROUP BY

count, sum, avg, min, max, the quantiles and the approximate aggregates commute, so they span HOP fine.

Scalar functions

Scalar functions (arithmetic, string, date and time, CAST) are DuckDB's, vendored whole, so they behave exactly as DuckDB's function reference documents them. The one that comes up on a stream is the timestamp cast a window needs: TUMBLE(CAST(ts AS TIMESTAMP), ...) for a string column, or epoch_ms(ts) for an integer epoch.