Data Filtering and Transformation

Business Scenario

The most common class of need is per-row processing — filter, convert, and grade each reading as it arrives, one in / one out synchronously, with no window aggregation. For example: discard obviously bogus readings (temperature out of range), convert Celsius to Fahrenheit, tag a level by temperature (normal/warning/alert). Such needs use non-aggregation mode with EmitSync, where the caller gets the single-row result directly.

SQL

SELECT deviceId,
       temperature,
       temperature * 1.8 + 32           AS temp_f,
       CASE WHEN temperature > 35 THEN 'CRITICAL'
            WHEN temperature > 30 THEN 'WARNING'
            ELSE 'OK' END               AS level
FROM stream
WHERE temperature > 0 AND temperature < 100

Input

{"deviceId": "dev-01", "temperature": 28.0}
{"deviceId": "dev-02", "temperature": 32.0}
{"deviceId": "dev-03", "temperature": 38.0}
{"deviceId": "dev-04", "temperature": 999.0}
{"deviceId": "dev-05", "temperature": null}

Output

EmitSync returns synchronously per row; rows that miss WHERE return nil (not an error):

{deviceId:dev-01 temperature:28   temp_f:82.4  level:OK}
{deviceId:dev-02 temperature:32   temp_f:89.6  level:WARNING}
{deviceId:dev-03 temperature:38   temp_f:100.4 level:CRITICAL}
[filtered] dev-04   // 999 out of range, caught by the bounds check
[filtered] dev-05   // null, caught by IS NOT NULL

Behavior Notes

• EmitSync is only for non-aggregation queries (no GROUP BY/window); calling it on an aggregation query errors with synchronous mode only supports non-aggregation queries.
• WHERE filters raw rows: rows that fail the condition return nil, so you can distinguish "filtered" from "error". To filter after aggregation, use HAVING (see Sliding Window and Continuous Detection).
• Arithmetic and aliases: temperature * 1.8 + 32 AS temp_f — write the expression inline and alias it, no UDF needed.
• CASE WHEN: for grading/conditional assignment; WHERE supports comparisons (>, <, =) and AND/OR to combine conditions.

How to run

Non-aggregation queries use EmitSync(data) for a synchronous single-row result; no AddSink needed. See How to Run Case SQL for the full boilerplate.

📚 Related Docs

• How to Run Case SQL
• SQL Reference — WHERE, expressions, CASE
• IoT Temperature Alerting and Metrics Aggregation — upgrading from single-point transform to windowed aggregation