Analytic Functions

Analytic functions use state to do analytic jobs. In streaming processing, analytic functions are evaluated first so that they are not affected by predicates in WHERE clause.

Analytic function call format is as below, where over clause is optional

AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>] [WHEN <Expression> [UNTIL <Expression>]])

Analytic function computations are performed over all the input events of the current query input, optionally you can limit analytic function to only consider events that match the partition_by_clause.

The syntax is like:

AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>])

The analysis function can use the WHEN clause to determine whether the current event is a valid event based on whether the condition is met. When it is a valid event, calculate the result and update the state according to the analysis function semantics. When it is an invalid event, ignore the event value and reuse the saved state value.

AnalyticFuncName(<arguments>...) OVER ([WHEN <Expression>])

LAG

lag(expr, [offset], [default value], [ignore null])

Returns the previous result of the expression at the specified offset.

Parameters:

• expr: The expression to evaluate
• offset (optional): Number of qualifying values to look back (default: 1). A value qualifies when its row satisfies WHEN, if present, and, if ignore null is true, the value is not null.
• default_value (optional): Value returned when no row is found at offset (default: nil)
• ignore_null (optional): Whether to ignore null values when looking back (default: true)

Behavior:

• With WHEN, lag(expr, 1) returns the most recent qualifying value and lag(expr, 2) returns the second most recent qualifying value. Rows that do not satisfy WHEN do not consume the offset.
• If no qualifying value exists at the specified offset, returns the default value
• If no default value is specified, returns nil
• When neither offset nor default value are specified, uses offset=1 and default=nil

Example function call to get the previous temperature value:

lag(temperature)

Example function call to get the previous temperature value with the same device id:

lag(temperature) OVER (PARTITION BY deviceId)

Example function call to calculate duration of events: ts is timestamp, and statusCode1 and statusCode2 are device status in the same event

select lag(Status) as Status, ts - lag(ts, 1, ts, true) OVER (WHEN had_changed(true, statusCode)) as duration from demo

LEAD

lead(expr, [offset], [default value], [ignore null])
  OVER ([PARTITION BY <partition key>] [WHEN <Expression> [UNTIL <Expression>]])

Returns the result of expr from a later input row. offset defaults to 1, default value defaults to nil, and ignore null defaults to true, matching lag. The offset counts qualifying future values: a value qualifies when its row satisfies WHEN, if present, and, if ignore null is true, the value is not null. For example, lead(expr, 2) OVER (WHEN condition) returns the second future qualifying value; rows that do not satisfy WHEN do not consume the offset. Because the result depends on future input, the current row is buffered until the requested future value is found, UNTIL becomes true, or the input ends.

WHEN selects future candidate rows. The offset is a successful-match condition, not a bound on how long or how many input rows LEAD may wait. UNTIL provides that separate stop condition. It is an eKuiper extension and is valid only together with WHEN; it is evaluated independently for every buffered row before WHEN. Within UNTIL, ordinary fields refer to the newly arrived probe row and current_row(expr) evaluates expr against the buffered origin row. If UNTIL is true, that request returns its default value. current_row is valid only in this context.

lead(candidate_t2) OVER (
  WHEN isNull(b) = false
  UNTIL ts - current_row(ts) > 5
)

UNTIL is data-driven and is checked only when input arrives. It does not create a processing-time timer or event-time watermark. A timer-driven time limit belongs to future WITHIN semantics.

For event-time rules, LEAD holds downstream watermarks behind buffered rows so that windows cannot close before those rows arrive. Watermarks can advance with subsequent input after the rows are released.

WHEN and the candidate expression are evaluated only if a pending request still needs a candidate after checking UNTIL. If evaluating a probe fails, none of that probe's LEAD decisions are committed and the probe is not added to the pending queue; later valid input can continue resolving existing requests.

Best practices

• Prefer an explicit UNTIL condition when a future match is not guaranteed, especially with selective WHEN conditions. Use WHEN true if every future row is a candidate but a stop condition is still needed.
• Choose a stop condition that keeps the number of pending requests small under the expected input rate. For example, with numeric ts values in milliseconds, UNTIL ts - current_row(ts) > 1000 stops waiting after a probe exceeds one second from the origin. This is a data-driven limit, not a timer or a hard buffer-size limit.
• Size the wait for each partition using approximately input rows per second × average wait in seconds. Even a short time interval can accumulate many requests at high input rates. Prefer simple conditions and verify them at the expected peak load.
• UNTIL is checked only when another row arrives in the same partition. An idle partition cannot expire its requests by itself. Output preserves global input order, so one unresolved early row can also hold back completed rows from other partitions.

Each probe checks the outstanding requests in its partition. Longer queues increase CPU and memory usage; adding UNTIL helps only if it actually keeps those queues short. Checkpoint snapshots also grow with the buffered state.

LATEST

latest(expr, [default value])

Return the latest non-null value of the expression. If not found, return the default value specified, if default value not set, return nil.

CHANGED_COL

changed_col(true, col)

Return the column value if it has changed from the last execution.

HAD_CHANGED

had_changed(true, expr1, expr2, ...)

Return if any of the columns had changed since the last run. The expression could be * to easily detect the change status of all columns.

Functions to detect changes

Changed_col function

This function is a normal scalar function, so it can be used in any clause including SELECT and WHERE.

Syntax

CHANGED_COL(<ignoreNull>, <expr>)

Arguments

ignoreNull: whether to ignore null values when comparing for changes. If true, the null value won’t emit a change.

expr: An expression to be selected and monitored for the changed status.

Returns

Return the changed value or nil with column name changed_col by default like any other functions. Use as alias to rename the column.

Changed_cols function

This function returns multiple columns, so it is only allowed in the SELECT clause.

Syntax

CHANGED_COLS (<prefix>, <ignoreNull>, <expr> [,...,<exprN>])

Arguments

prefix: The prefix of the selected column name. By default, the selected name will be the same as select the expr directly. For example, CHANGED_COLS("", true, col1) will return col1 as the name. If setting a prefix, the return name will have that prefix. For example, CHANGED_COLS("changed_", true, col1) will return changed_col1 as the name.

ignoreNull: whether to ignore null values when detecting changes. If true, the null value won’t trigger a change.

expr: An expression to be selected and monitored for the changed status. Allow any expression that can be used in select clause. The expression can be a * which will return multiple columns by one expression.

Returns

Return all changed values compared to the previous sink result. So if used in a scalar rule, it will compare to the previous value emitting. If used in a window, it will compare to the previous window result.

In the first run, all expressions will be returned because there is no previous result.

In the consequent runs, if nothing changed, it can emit nothing. And if the sink has the default omitEmpty, the sink will not be triggerred.

Notice

The multiple column outputs can only be used in the select clause. Even the selected result cannot be accessed in WHERE or other place. If a filter based on the value is needed, use CHANGED_COL or set the result of multiple column outputs as the prior rule in a rule chain.

For multiple column outputs, the alias can only be set generally with the prefix. To set alias for each column separately, try to call the changed function for each column respectively and use as to set alias.

Had_changed function

This function is a scalar function with one or more arguments.

HAD_CHANGED (<ignoreNull>, <expr> [,...,<exprN>])

Arguments

ignoreNull: whether to ignore null values when detecting changes. If true, the null value won’t trigger a change.

expr: An expression to be monitored for the changed status. Allow any expression that can be used in select clause. The expression can be a * to detect changes of all columns easily.

Returns

Return a bool value to indicate the changed status if any of the arguments had changed since the last run. The multiple arguments' version is a handy way to check HAD_CHANGED(expr1) OR HAD_CHANGED(expr2) ... OR HAD_CHANGED(exprN). To detect another relationship, just use separate HAD_CHANGED functions. For example, to check if all expressions are changed HAD_CHANGED(expr1) AND HAD_CHANGED(expr2) ... AND HAD_CHANGED(exprN).

Examples

Create a stream demo and have below inputs

{
  "ts": 1,
  "temperature": 23,
  "humidity": 88
}
{
  "ts": 2,
  "temperature": 23,
  "humidity": 88
}
{
  "ts": 3,
  "temperature": 23,
  "humidity": 88
}
{
  "ts": 4,
  "temperature": 25,
  "humidity": 88
}
{
  "ts": 5,
  "temperature": 25,
  "humidity": 90
}
{
  "ts": 6,
  "temperature": 25,
  "humidity": 91
}
{
  "ts": 7,
  "temperature": 25,
  "humidity": 91
}
{
  "ts": 8,
  "temperature": 25,
  "humidity": 91
}

Rule to get the changed temperature values:

SQL: SELECT CHANGED_COLS("", true, temperature) FROM demo
___________________________________________________
{"temperature":23}
{"temperature":25}

Rule to get the changed temperature and humidity values, and rename the changed value in a unified prefix:

SQL: SELECT CHANGED_COLS("c_", true, temperature, humidity) FROM demo
_________________________________________________________
{"c_ts":1, "c_temperature":23, "c_humidity":88}
{"c_ts":2}
{"c_ts":3}
{"c_ts":4, "c_temperature":25}
{"c_ts":5, "c_humidity":90}
{"c_ts":6, "c_humidity":91}
{"c_ts":7}
{"c_ts":8}

Rule to get the changed values of all columns and do not ignore null:

SQL: SELECT CHANGED_COLS("c_", false, *) FROM demo
_________________________________________________________
{"c_temperature":23,"c_humidity":88}
{"c_temperature":25}
{"c_humidity":90}
{"c_humidity":91}

Rule to get the average value change in a window:

SQL: SELECT CHANGED_COLS("t", true, avg(temperature)) FROM demo GROUP BY CountWindow(2)
_________________________________________________________________
{"tavg":23}
{"tavg":24}
{"tavg":25}

Rule to get the events when temperature or humidity changed:

SQL: SELECT ts, temperature, humidity FROM demo
WHERE HAD_CHANGED(true, temperature, humidity) = true
_________________________________________________________
{"ts":1,temperature":23,"humidity":88}
{"ts":4,temperature":25,"humidity":88}
{"ts":5,temperature":25,"humidity":90}
{"ts":6,temperature":25,"humidity":91}

Rule to get the events when temperature has changed but humidity has NOT changed:

SQL: SELECT ts, temperature, humidity FROM demo
WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
_________________________________________________________
{"ts":4,temperature":25,"humidity":88}

Rule to get the changed temperature and humidity value with customized names:

SQL: SELECT CHANGED_COL(true, temperature) AS myTemp, CHANGED_COL(true, humidity) AS myHum FROM demo
_________________________________________________________
{"myTemp":23,"myHum":88}
{"myTemp":25}
{"myHum":90}
{"myHum":91}

Rule to get the changed values when the temperature had changed to value bigger than 24:

SQL: SELECT ts, temperature, humidity FROM demo
WHERE CHANGED_COL(true, temperature) > 24
_________________________________________________________
{"ts":4,temperature":25,"humidity":88}

ACC Functions

The ACC Functions means the accumulate functions, which will perform cumulative calculations based on the obtained parameters, and the cumulative scope is the entire life cycle of the rule.

For the next acc functions, we will simulate input and output with the following data:

a

Enter 3 pieces of data in sequence, 1,2,3 respectively.

ACC_SUM

acc_sum(expr)

The acc_sum function accumulates the expression results and returns the cumulative sum result.

Example 1: Cumulative sums using acc_sum

acc_sum(a)

The results are: 1 3 6

ACC_MAX

acc_max(expr)

The acc_max function performs accumulative comparison on the result of the expression to take the larger value, and returns the result of the cumulative comparison to take the larger value.

Example 1: Use acc_max for cumulative comparison to take the larger value

acc_max(a)

The results are: 1 2 3

ACC_MIN

acc_min(expr)

The acc_min function performs accumulative comparison on the result of the expression to take the smaller value, and returns the result of the cumulative comparison to take the smaller value.

Example 1: Use acc_min for cumulative comparison to take the smaller value

acc_min(a)

The results are: 1 1 1

ACC_COUNT

acc_count(expr)

The acc_count function counts the cumulative number of expression results and returns the cumulative value.

Example 1: Use acc_count for cumulative count statistics

acc_count(a)

The results are: 1 2 3

ACC_AVG

acc_avg(expr)

The acc_avg function performs cumulative average statistics on the expression result and returns the cumulative average.

Example 1: Cumulative average statistics using acc_count

acc_avg(a)

The results are: 1 1.5 2

ACC_COLLECT

acc_collect(expr)

The acc_collect function collects non-nil expression results into an array, preserving insertion order.

Example 1: Collect values using acc_collect

acc_collect(a)

The results are: [1] [1,2] [1,2,3]

ACC_MAX_BY

acc_max_by(value, compare_value)

The acc_max_by function cumulatively compares compare_value and returns the value associated with its greatest value. If compare_value is equal, the value from the latest event is used. It returns nil when there is no valid compare_value.

Example: get the collection timestamp associated with the cumulative maximum temperature.

acc_max_by(ts, temp) over (partition by soc)

ACC_MIN_BY

acc_min_by(value, compare_value)

The acc_min_by function cumulatively compares compare_value and returns the value associated with its smallest value. If compare_value is equal, the value from the latest event is used. It returns nil when there is no valid compare_value.

Example: get the collection timestamp associated with the cumulative minimum temperature.

acc_min_by(ts, temp) over (partition by soc)

ACC_MAP_AGG

acc_map_agg(key, value)

The acc_map_agg function cumulatively builds a key-value array. The key is converted to a string. If a key is repeated, its value is updated with the latest value, while the first-seen key order is preserved.

Each returned item is an object containing key and value fields.

Example:

acc_map_agg(soc, object_construct(
    'max_temp', max_temp,
    'max_temp_ts', max_temp_ts
))

Example result:

[
  {"key": "18", "value": {"max_temp": 30, "max_temp_ts": 1788000060000}},
  {"key": "19", "value": {"max_temp": 31, "max_temp_ts": 1788000090000}}
]

ACC function with conditions

ACC function can define the starting point and reset point of cumulative calculation by accepting additional expression parameters. The specific usage is as follows

acc_count(a,expr1,expr2)

Where expr1 represents the starting point of cumulative calculation, and expr2 represents the reset point of cumulative calculation.

Example: Use acc_count to perform cumulative count with conditions

acc_count(a, a > 1, a < 0)

The following data are obtained:

a = 1
a = 2
a = 1
a = 3
a = -1
a = 1

The results are as follows:

0
1
2
3
4
0