StreamSQL Analytical Functions

Analytical functions perform cross-event stateful computations on a continuous event stream — e.g. "previous value", "has it changed", "running sum". Each event is evaluated immediately upon arrival, with state retained across events. They are usable in SELECT and WHERE, and are typically used for change detection (CDC).

An optional OVER clause controls partitioning and conditional state:

func(args) OVER ( [PARTITION BY col[, col...]] [WHEN condition] )

• PARTITION BY: maintains independent state per partition (e.g. "previous value per device").
• WHEN condition: only events satisfying the condition update state; others reuse the last result.
• ORDER BY / ROWS BETWEEN are not supported (streamsql's analytic functions are per-event state machines, not Flink-style window frames).

Analytic functions cannot appear in HAVING; for threshold / sustained detection use windowed aggregation + HAVING.

Function List

lag — Previous value

Syntax: lag(field [, offset [, default [, ignoreNull]]])
Description: Returns the value from the offset-th event before the current row (default offset=1; returns default when there are not enough rows).
Example:

SELECT temperature, lag(temperature) AS prev FROM stream

latest — Latest non-null value

Syntax: latest(field [, default])
Description: Returns the latest non-null value of the field (nil does not update state).
Example:

SELECT latest(temperature) AS lt FROM stream

had_changed — Whether a value changed

Syntax: had_changed(ignoreNull, field[, field...])
Description: Returns a boolean: whether any value changed vs. the previous row (the first row counts as changed). Supports multiple fields; "*" checks every column of the row.
Example:

SELECT ts FROM stream WHERE had_changed(true, status) == true

changed_col — Changed column value (single-column scalar)

Syntax: changed_col(ignoreNull, field)
Description: Returns the new value when changed, nil when unchanged (omitted from projection).
Example:

SELECT changed_col(true, temperature) AS chg FROM stream

changed_cols — Multiple changed column values (dynamic columns)

Syntax: changed_cols(prefix, ignoreNull, field[, field...])
Description: Returns {prefix+column: new value} containing only changed columns; "*" checks all columns of the row. SELECT only.
Example:

SELECT changed_cols("c_", true, temperature, humidity) FROM stream

acc_sum / acc_max / acc_min / acc_count / acc_avg — Lifecycle accumulation

Syntax: acc_sum(field [, startExpr [, resetExpr]]) (likewise for the other acc_*)
Description: Lifecycle accumulation across events (sum / max-min / count / avg); optional startExpr / resetExpr for conditional start and reset.
Example:

SELECT acc_sum(power) AS total_power FROM stream

📚 Related Documentation

• Aggregate Functions - Learn detailed usage of aggregate functions
• Window Functions - Learn detailed usage of window functions
• SQL Reference - View complete SQL syntax reference