streamTransform

Node Type: x/streamTransform

Description: Stream transformer node, based on the StreamSQL engine, uses SQL syntax to filter, transform, and process fields of real-time data streams. It processes non-aggregate queries synchronously per event (state is preserved across events), supporting data filtering, field transformation, format conversion, change detection, and lifetime accumulation. Supports single data and array data input.

Upgrade Notice (v0.37.0)

Filtered rows (WHERE not matched, or changed_cols/changed_col unchanged) now route to the False chain instead of Failure (filtering is not an error). If your DSL wired the filtered output to the Failure chain, reconnect it to the False chain; Failure is reserved for genuine processing errors.

Features

• SQL Syntax: Use standard SQL syntax for data transformation, low learning cost.
• Real-time Processing: Synchronous processing of single data and array data.
• Field Operations: Supports field selection, renaming, calculation, and conditional filtering.
• Function Support: 60+ built-in functions, including math, string, time, etc.
• Analytic Functions: Cross-event stateful computation — lag/latest, had_changed/changed_col/changed_cols, acc_* (see Analytic Functions).
• Stream-Table JOIN: Enrich stream rows with metadata tables (see SQL Reference).
• Conditional Filtering: Supports WHERE clause for data filtering.
• Array Processing: Automatically processes array data, transforms each element and merges the results.

Input Data Support

This node supports two input data formats:

Single Data Input

Directly process a single JSON object: successful transformation goes to Success; WHERE not matched goes to False; a processing error goes to Failure:

{"deviceId": "sensor001", "temperature": 25.5, "humidity": 60.2}

Array Data Input

Automatically process JSON arrays, traverse each element for transformation, and merge the successfully transformed results into a new array for output:

[
  {"deviceId": "sensor001", "temperature": 25.5, "humidity": 60.2},
  {"deviceId": "sensor002", "temperature": 28.3, "humidity": 55.8},
  {"deviceId": "sensor003", "temperature": 22.1, "humidity": 65.4}
]

Array Processing Description

• Each element in the array will be processed by SQL transformation one by one.
• Only elements that are successfully transformed and meet the WHERE condition will be included in the output array.
• If at least one element is successfully transformed, the merged array is output through the Success chain.
• If all elements are filtered by WHERE (no errors), the merged result goes to False; if any element errors, it goes to Failure.
• Message metadata will contain processing statistics: originalCount, transformedCount, failedCount.

Configuration

Field	Type	Description	Default Value
sql	string	Transformation SQL statement, must be a non-aggregate query (cannot contain GROUP BY / aggregation functions; analytic functions are allowed)	None
tables	array	Optional, metadata table configuration for stream-table JOIN enrichment (see below)	None

SQL Syntax Support

Detailed Syntax Reference

For complete SQL syntax instructions, please refer to: StreamSQL SQL Syntax Reference

Relation Types

• Success: Successful transformation; the transformed data is passed through this relation chain (metadata match=true).
• False: Data was filtered (WHERE not matched, or changed_cols/changed_col had no change) — not an error; metadata match=false.
• Failure: A processing error occurred (non-JSON data, SQL evaluation error, etc.); error information is passed through this relation chain.

Distinguishing "filtered" from "error"

Being filtered by WHERE or having no changed_cols change is an intentional no-output, routed to False. Only genuine processing errors go to Failure. So the Failure chain can safely be wired to alerts/error handling without being polluted by normal event compression.

Execution Results

Success Chain Output

Transformed data, the format is determined by the SQL query result:

{
  "field1": "transformed_value1",
  "field2": "transformed_value2",
  "calculated_field": 123.45
}

False Chain Output

The original (untransformed) input data: when WHERE is not matched or changed_cols/changed_col has no change, the message body is unchanged — still the original incoming data (metadata match=false), so downstream can inspect/log what was filtered.

Failure Chain Output

Error message, containing specific error descriptions.

Configuration Examples

Basic Field Transformation

{
  "id": "s1",
  "type": "x/streamTransform",
  "name": "Temperature Unit Conversion",
  "configuration": {
    "sql": "SELECT deviceId, temperature, humidity, temperature * 1.8 + 32 as temp_fahrenheit FROM stream WHERE temperature IS NOT NULL",
    "debug": false
  }
}

Data Filtering and Calculation

{
  "id": "s2",
  "type": "x/streamTransform",
  "name": "High Temperature Data Processing",
  "configuration": {
    "sql": "SELECT deviceId, temperature, CASE WHEN temperature > 30 THEN 'HIGH' WHEN temperature < 10 THEN 'LOW' ELSE 'NORMAL' END as temp_level FROM stream WHERE temperature > 20",
    "debug": true
  }
}

String Processing

{
  "id": "s3",
  "type": "x/streamTransform",
  "name": "Device Information Formatting",
  "configuration": {
    "sql": "SELECT UPPER(deviceId) as device_id, CONCAT(location, '-', deviceType) as device_info, ROUND(temperature, 2) as temp FROM stream",
    "debug": false
  }
}

Analytic Functions (Cross-Event State)

streamTransform processes each event synchronously, making it a natural fit for analytic functions that preserve state across events — each message is evaluated on arrival, and state is retained across messages for the node instance's lifetime.

-- CDC change detection: output when current crosses 300A from below
SELECT current, deviceId FROM stream
WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300

-- Output only when temperature changes
SELECT ts, temperature FROM stream WHERE had_changed(true, temperature) == true

-- Send only the changed fields, with a c_ prefix
SELECT changed_cols("c_", true, temperature, humidity) FROM stream

-- Lifetime accumulation (not reset by windows)
SELECT acc_sum(power) AS total, acc_count(*) AS cnt FROM stream

"No change" goes to the False chain

When changed_cols/changed_col is the sole output and nothing changed this time, the node returns nil → routed to the False chain (match=false). This is the intended event compression — wire the False chain to drop/ignore; it never pollutes the Failure chain.

For full syntax and more usage (OVER/PARTITION BY/WHEN, conditional accumulation, etc.) see Analytic Functions.

Stream-Table JOIN (Metadata Enrichment)

Configure tables to load metadata tables (device→location, product→category, etc.) and JOIN them in SQL to enrich stream rows. Tables can be loaded from file/http and refreshed periodically; the JOIN index key is auto-derived from the ON clause.

{
  "id": "enrich", "type": "x/streamTransform", "name": "Device Enrichment",
  "configuration": {
    "sql": "SELECT deviceId, m.location, m.type FROM stream s LEFT JOIN meta m ON s.deviceId = m.deviceId WHERE s.temp > 30",
    "tables": [
      {"name": "meta", "source": "file", "path": "/etc/rulego/device_meta.json", "format": "json", "refresh": "30s"}
    ]
  }
}

For the full tables fields and JOIN syntax see SQL Reference.

Application Examples

Example 1: IoT Data Preprocessing

Scenario: Clean and format raw data reported by IoT devices.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "iot_data_preprocessing",
    "name": "IoT Data Preprocessing",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Data Cleaning",
        "configuration": {
          "sql": "SELECT deviceId, temperature, humidity, pressure, CASE WHEN temperature > 50 OR temperature < -20 THEN 'INVALID' ELSE 'VALID' END as data_quality FROM stream WHERE deviceId IS NOT NULL"
        }
      },
      {
        "id": "s2",
        "type": "jsFilter",
        "name": "Effective Data Filtering",
        "configuration": {
          "jsScript": "return msg.data_quality === 'VALID';"
        }
      },
      {
        "id": "s3",
        "type": "x/streamTransform",
        "name": "Unit Conversion",
        "configuration": {
          "sql": "SELECT deviceId, ROUND(temperature, 2) as temperature_c, ROUND(temperature * 1.8 + 32, 2) as temperature_f, ROUND(humidity, 1) as humidity_percent, pressure FROM stream"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "Processing Result",
        "configuration": {
          "jsScript": "return 'Processed: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s5",
        "type": "log",
        "name": "Invalid Data",
        "configuration": {
          "jsScript": "return 'Invalid data: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "Success"},
      {"fromId": "s2", "toId": "s3", "type": "True"},
      {"fromId": "s2", "toId": "s5", "type": "False"},
      {"fromId": "s3", "toId": "s4", "type": "Success"}
    ]
  }
}

Input Data:

{"deviceId": "sensor001", "temperature": 25.678, "humidity": 65.432, "pressure": 1013.25}

Output Result:

{
  "deviceId": "sensor001",
  "temperature_c": 25.68,
  "temperature_f": 78.22,
  "humidity_percent": 65.4,
  "pressure": 1013.25
}

Example 2: Data Standardization

Scenario: Standardize device data in different formats into a unified format.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "data_standardization",
    "name": "Data Standardization",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Field Standardization",
        "configuration": {
          "sql": "SELECT UPPER(COALESCE(device_id, deviceId, id)) as device_id, COALESCE(temp, temperature, t) as temperature, COALESCE(hum, humidity, h) as humidity, CONCAT(COALESCE(location, 'unknown'), '-', COALESCE(building, 'default')) as location_info FROM stream"
        }
      },
      {
        "id": "s2",
        "type": "x/streamTransform",
        "name": "Data Classification",
        "configuration": {
          "sql": "SELECT *, CASE WHEN temperature > 25 AND humidity > 60 THEN 'HOT_HUMID' WHEN temperature > 25 THEN 'HOT_DRY' WHEN humidity > 60 THEN 'COOL_HUMID' ELSE 'COMFORTABLE' END as environment_type FROM stream"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Standardization Result",
        "configuration": {
          "jsScript": "return 'Standardized: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "Success"},
      {"fromId": "s2", "toId": "s3", "type": "Success"}
    ]
  }
}

Example 3: Batch Processing of Array Data

Scenario: Process an array message containing multiple sensors' data for batch temperature conversion and filtering.

Input Data:

[
  {"sensorId": "s001", "value": 23.5, "unit": "C", "status": "active"},
  {"sensorId": "s002", "value": 45.2, "unit": "C", "status": "active"},
  {"sensorId": "s003", "value": 18.7, "unit": "C", "status": "inactive"},
  {"sensorId": "s004", "value": 35.8, "unit": "C", "status": "active"}
]

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "batch_transform",
    "name": "Batch Data Transformation",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Batch Temperature Conversion",
        "configuration": {
          "sql": "SELECT sensorId, value as celsius, ROUND(value * 1.8 + 32, 1) as fahrenheit, CASE WHEN value > 30 THEN 'HIGH' ELSE 'NORMAL' END as temp_status FROM stream WHERE status = 'active'"
        }
      },
      {
        "id": "s2",
        "type": "log",
        "name": "Transform Success",
        "configuration": {
          "jsScript": "return 'Transformed ' + metadata.getValue('transformedCount') + ' out of ' + metadata.getValue('originalCount') + ' items: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Transform Failure",
        "configuration": {
          "jsScript": "return 'Failed to transform: ' + metadata.getValue('failedCount') + ' out of ' + metadata.getValue('originalCount') + ' items';"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "Success"},
      {"fromId": "s1", "toId": "s3", "type": "Failure"}
    ]
  }
}

Success Chain Output:

[
  {"sensorId": "s001", "celsius": 23.5, "fahrenheit": 74.3, "temp_status": "NORMAL"},
  {"sensorId": "s002", "celsius": 45.2, "fahrenheit": 113.4, "temp_status": "HIGH"},
  {"sensorId": "s004", "celsius": 35.8, "fahrenheit": 96.4, "temp_status": "HIGH"}
]

Message Metadata:

{
  "match": "true",
  "originalCount": "4",
  "transformedCount": "3",
  "failedCount": "1"
}

Processing Notes

• The original array contains 4 elements.
• The WHERE condition filtered out sensor s003 with status 'inactive'.
• The final output array contains 3 successfully transformed elements.
• Metadata records detailed processing statistics.

Example 4: CDC Change Detection (Current Crossing Threshold)

Scenario: Output when a device's current goes from low to high and crosses 300A (current > 300 and the previous value < 300).

Node Configuration:

{
  "id": "s4",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT current, deviceId, ts FROM stream WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300"
  }
}

Input/Output:

{current:200, deviceId:1, ts:2}
{current:500, deviceId:1, ts:3}  → output (200→500 crosses threshold)
{current:600, deviceId:1, ts:4}  → not output (500→600 does not cross)

PARTITION BY deviceId gives each device its own "previous current"; lag is evaluated before WHERE so that WHERE can reference it.

Example 5: Event Compression (Send Only Changed Fields)

Scenario: Upstream bandwidth is limited; each message carries only the fields that changed, with a c_ prefix.

Node Configuration:

{
  "id": "s5",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT changed_cols(\"c_\", true, temperature, humidity) FROM stream"
  }
}

Input/Output:

{temperature:23, humidity:50} → {c_temperature:23, c_humidity:50} (first time, all change)
{temperature:23, humidity:55} → {c_humidity:55} (only humidity changes)
{temperature:23, humidity:55} → goes to the False chain (nothing changed, row suppressed)

No change goes to the False chain

When changed_cols is the sole output and nothing changed, it returns nil → routed to the False chain (match=false). This is the intended event compression — wire the False chain to drop/ignore.

Example 6: Lifetime Accumulation (Statistics Since Boot)

Scenario: Track total energy, peak, and sample count since boot — not reset by windows, accumulated per event.

Node Configuration:

{
  "id": "s6",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT acc_sum(power) AS total, acc_max(power) AS peak, acc_count(*) AS cnt, acc_avg(power) AS avg_power FROM stream"
  }
}

Each event accumulates on arrival and outputs the current accumulated value; state is retained for the node's lifetime.

Notes

1. SQL Syntax Restriction: Only non-aggregate queries are supported; cannot contain GROUP BY / aggregation functions (COUNT/SUM/AVG, etc.); analytic functions (lag/changed_col/acc_*, etc.) are not aggregations and are allowed.
2. Data Type: Only JSON data type input is supported.
3. Synchronous Processing: Transformation is synchronous and blocks the current message.
4. Array Processing:
   • Each element in the array is processed by SQL transformation one by one.
   • Only elements that are successfully transformed and meet the WHERE condition are included in the output array.
   • Partial element failures do not affect the overall result, only the final array element count.
   • Metadata automatically adds processing statistics: originalCount, transformedCount, failedCount.
5. WHERE Condition: Data that does not meet the WHERE condition is filtered out (routed to the False chain) and excluded from the output.
6. Performance: For large arrays, consider the impact of data volume on processing performance.