Change Data Capture (CDC) Case Study

Case Overview

In IoT, "change data capture" means identifying events that actually changed from a continuous device-report stream — a metric jumping across a threshold, a status flipping, a field changing, etc. StreamSQL's analytic functions (lag / had_changed / changed_col / changed_cols / latest / acc_*) are designed precisely for this per-event state computation: each datum is evaluated on arrival, state persists across events, and they can be used in both SELECT and WHERE.

The cases in this set are all IoT scenarios; the SQL and outputs have been run and verified against StreamSQL.

Demo Data

The three "current rises across the threshold" cases below share this simulated input (two devices reporting current alternately):

{"current": 300, "deviceId": 1, "ts": 1}
{"current": 400, "deviceId": 2, "ts": 2}
{"current": 200, "deviceId": 1, "ts": 3}
{"current": 200, "deviceId": 2, "ts": 4}
{"current": 500, "deviceId": 1, "ts": 5}
{"current": 200, "deviceId": 2, "ts": 6}
{"current": 400, "deviceId": 1, "ts": 7}
{"current": 600, "deviceId": 2, "ts": 8}

Current Rises Across the Threshold (lag)

A plain current > 300 will alert repeatedly while current stays high. What you actually want is "the moment it goes from at-or-below the threshold to above it" — which implicitly means "compare with the previous value", exactly what lag is for.

Case 1: Global — the previous value across the whole stream

Without splitting by device, treat the whole stream as one sequence and track the "previous value".

SELECT current, ts
FROM stream
WHERE current > 300 AND lag(current) <= 300

lag(current) has no OVER; state is shared across all events. Output (the moments current jumps from ≤300 to >300):

{"current": 400, "ts": 2}
{"current": 500, "ts": 5}
{"current": 400, "ts": 7}

Note

This rule detects change across the whole stream (all devices mixed in). For example, the previous value at ts5 is 200 from ts4 (device 2), not device 1's own previous value. To split by device, see Case 2.

Case 2: Partition by device — each device's own previous value

PARTITION BY deviceId lets each device keep its own "previous value", independent of the others.

SELECT current, deviceId, ts
FROM stream
WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300

Output (the moment each device individually jumps from <300 to >300):

{"current": 500, "deviceId": 1, "ts": 5}
{"current": 600, "deviceId": 2, "ts": 8}

Case 3: Watch a single device — OVER WHEN scopes the lag

We only care about deviceId = 1. Filter with WHERE deviceId = 1, then use OVER (WHEN deviceId = 1) so lag updates state only on that device's events, excluding other devices' data.

SELECT current, deviceId, ts
FROM stream
WHERE current > 300 AND deviceId = 1
  AND lag(current) OVER (WHEN deviceId = 1) < 300

Output:

{"current": 500, "deviceId": 1, "ts": 5}

OVER (WHEN ...) is the analytic function's conditional state semantics: only events that satisfy the condition update state; others reuse the last result. Besides lag, analytic functions like had_changed also support the OVER clause.

Other Change-Detection Capabilities

Case 4: Device status flip alert (had_changed)

Devices report a running status (running / fault / offline…). Output the moment the status changes; stay silent when it does not.

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

Given input running → running → fault → fault → running, it outputs the first row and every row where the status flips. ignoreNull=true keeps sensor-offline nils from polluting the baseline.

Case 5: Report only the changed sensor fields (changed_cols)

An edge gateway has limited uplink bandwidth; require each message to carry only the changed fields (with a prefix), and when nothing changed at all, do not send.

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

Given input (23,50) → (23,55) → (23,55) → (25,55), output:

{"c_temperature": 23, "c_humidity": 50}
{"c_humidity": 55}
(the third row is entirely unchanged, not emitted)
{"c_temperature": 25}

To detect/output "any column changed", swap the column names for "*": had_changed(true, "*"), changed_cols("d_", true, "*").

Case 6: Latest valid reading + accumulated energy (latest + acc_*)

Sensors occasionally go offline and report nil. latest always returns the latest non-null value; acc_* accumulates over the rule's lifetime.

SELECT deviceId,
       latest(temperature)   AS last_temp,
       acc_sum(power)        AS total_power,
       acc_max(power)        AS peak_power,
       acc_count(*)          AS sample_cnt
FROM stream

Contrast: Sustained Over-Threshold (Window + HAVING, Not Change Detection)

Note

"Above threshold for N consecutive seconds" is not change detection; do not use lag or over(when) on a window. Mainstream engines (Flink/Spark, etc.) all use window aggregation + HAVING.

Require bus current to be above 200A for the entire 10 seconds to alert (a dip in between does not count): a sliding window aggregates all events in those 10 seconds (including dips below 200), and HAVING mn > 200 filters out windows that contained a dip.

SELECT min(concurrency) AS mn, count(*) AS c
FROM stream
GROUP BY SlidingWindow(ss, 10)
HAVING mn > 200

Note that HAVING references the SELECT alias mn.

CDC Capability Map

Need	What to use	Example
Compare with previous / cross threshold	lag (+ OVER PARTITION BY/WHEN)	Current jumps across a threshold
Whether it changed	had_changed / changed_col	Status flip, single-field jump
Which fields changed (dynamic columns)	changed_cols	Edge bandwidth compression
Latest non-null value (nil-proof)	latest	Sensor goes offline
Lifetime accumulation	acc_sum/max/min/count/avg	Total energy, peak
Any column changed	had_changed(true, "*")	Whole-row change audit
Above threshold for N seconds	Window aggregation + HAVING (not analytic functions)	Sustained overload alert

📚 Related Docs

• Analytic Functions - Full analytic-function syntax and function list
• SQL Reference - HAVING, window functions, and the full SQL syntax
• IoT Temperature Alerting and Metrics Aggregation - Windowed-aggregation IoT cases