Streamsql

StreamSQL 流式计算组件

RuleGo 通过 x/streamAggregator 和 x/streamTransform 两个组件提供流式 SQL 处理能力。

组件选择

用错组件会初始化失败：aggregator 的 SQL 必须是聚合或 CEP（MATCH_RECOGNIZE），transform 不能含聚合函数/MATCH_RECOGNIZE。分析函数（lag/changed_col/acc_* 等）不是聚合函数，两边都可用——不带窗口时进 transform，带窗口时进 aggregator。

x/streamAggregator 流聚合器

数据流：输入 → 加入聚合流（原始数据走 Success 继续）→ 窗口触发 → 聚合结果走 stream_event。

{
  "type": "x/streamAggregator",
  "configuration": {
    "sql": "SELECT deviceId, AVG(temperature) as avg_temp, COUNT(*) as count FROM stream GROUP BY deviceId, TumblingWindow('30s')"
  }
}

连接类型：Success（原始数据）、Failure（非 JSON/SQL 错误）、stream_event（聚合结果，关键）。

结果消息：msgType=stream_event，metadata.queryType=aggregation(聚合)或 cep(MATCH_RECOGNIZE)、resultType=window_triggered/pattern_matched，msg.data 是结果 JSON 数组，如 [{"deviceId":"d1","avg_temp":25.5,"count":10}]。可同时从 Success（原始）和 stream_event（结果）双路输出分别处理。

x/streamTransform 流转换器

{
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT deviceId, temperature * 1.8 + 32 as fahrenheit FROM stream WHERE temperature > 20"
  }
}

连接类型：Success（转换成功且符合 WHERE，结果替换 msg.data，match=true）、False（WHERE 不命中 / changed_cols 无变化，携带原始数据，match=false）、Failure（错误）。

数组输入逐条转换后合并输出，metadata 额外含 originalCount/transformedCount/failedCount。

分析函数（跨事件状态）

逐条事件求值、状态跨事件保留，走 transform 的同步路径；窗口内则对窗口输出求值、状态跨窗口保留。

-- CDC：电流从低跨过 300A
SELECT current, deviceId FROM stream
WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300
-- 仅发送变化字段
SELECT changed_cols("c_", true, temperature, humidity) FROM stream
-- 窗口均值变化才输出（进 aggregator）
SELECT changed_cols("t", true, avg(temperature)) FROM stream GROUP BY CountingWindow(2)

• OVER (PARTITION BY ... WHEN ...)：分区 / 条件更新；不支持 ORDER BY/ROWS frame。
• acc_count(v, startExpr, resetExpr)：条件累计（开始点 / 重置点）。
• 条件表达式用比较运算符 >/</==；分析函数参数里 = 不作相等判断，字符串相等用 ==。
• changed_cols/changed_col 唯一输出且无变化时返回 nil → 走 transform 的 False 链（预期的事件压缩，接丢弃/忽略即可）。

CEP 模式识别（MATCH_RECOGNIZE）

x/streamAggregator 支持 MATCH_RECOGNIZE，语法对齐 Flink / SQL 标准的行模式识别（标准语法详见 Flink MATCH_RECOGNIZE 文档）。匹配命中后结果走 stream_event（与聚合同出口，metadata.queryType=cep、resultType=pattern_matched）。

SELECT * FROM stream
MATCH_RECOGNIZE (
  PARTITION BY deviceId                         -- 可选：分区键
  ORDER BY ts                                   -- 必填：排序字段
  MEASURES MATCH_NUMBER() AS mn, A.v AS peak    -- 输出列；可含聚合 COUNT(*)/SUM(A.v)/AVG(v)/MAX(v)，及 FIRST(v)/LAST(v)/CLASSIFIER()
  ONE ROW PER MATCH                             -- 默认；或 ALL ROWS PER MATCH
  AFTER MATCH SKIP PAST LAST ROW                -- 可选：PAST LAST ROW(默认)/TO NEXT ROW/TO FIRST x/TO LAST x
  PATTERN (A{3} B?)                             -- 拼接 AB、交替 A|B、量词 {n}/{n,m}/{n,}/*/+/?、分组 ()、PERMUTE(A,B)
  WITHIN '1h'                                   -- 可选：时间窗（默认 1h）
  DEFINE A AS v > 50, B AS v < 10               -- 符号条件；未在 DEFINE 出现的符号恒为真
)

已支持：PARTITION BY、ORDER BY（必填）、MEASURES（含聚合 + MATCH_NUMBER()）、ONE/ALL ROWS PER MATCH、AFTER MATCH SKIP 全部策略、PATTERN（拼接 / 交替 / 量词 / 分组 / PERMUTE）、SUBSET、WITHIN、DEFINE。

不支持 / 注意：

• {- ... -} 排除（absence）不支持，会报错。
• 仅单流（FROM stream）；不支持多流 MATCH_RECOGNIZE。
• 量词默认贪婪；懒惰量词（*?/+?）亦支持，但同一模式内勿混用贪婪与懒惰。
• ⚠️ MATCH_RECOGNIZE 子句本身的语法错误会被外层 Parse 容错吞掉，导致查询静默降级为非 CEP（不报错）。DEFINE/MEASURES 里的表达式语法错会在 Execute 期暴露。写完建议实跑验证或用 IsCEPQuery() 确认走了 CEP 路径。

SQL 语法

SELECT [DISTINCT] select_list
FROM stream
[WHERE condition]
[GROUP BY grouping_element [, ...]]
[HAVING condition]
[ORDER BY ordering_list]
[LIMIT count]
[WITH (option = value [, ...])]

• SELECT：字段 / * / 计算字段 / 别名（AS 可省）/ DISTINCT / 聚合函数。
• 嵌套字段：点号、[index]（支持负数）、['key']，如 device.info.name、items[0].name、config['host']。可用于 WHERE/GROUP BY/HAVING。
• CASE：CASE WHEN cond THEN v ... ELSE d END，可在聚合内部做条件计数 SUM(CASE WHEN ... THEN 1 ELSE 0 END)。
• WHERE：比较 / AND OR NOT / BETWEEN / IN / IS [NOT] NULL / LIKE（%/_），均可用嵌套字段。
• HAVING：过滤聚合结果。引用 SELECT 别名（HAVING avg_temp > 25），不能复述聚合函数。

窗口函数（aggregator 必含一个）

GROUP BY TumblingWindow('30s')                                        -- 滚动：固定大小、无重叠
GROUP BY SlidingWindow('5m', '1m')                                    -- 滑动：大小 + 滑动间隔
GROUP BY CountingWindow(100)                                          -- 计数：按条数触发
GROUP BY user_id, SessionWindow('5m')                                 -- 会话：超时关闭
GROUP BY deviceId, GLOBAL WINDOW TRIGGER WHEN COUNT(*) >= 3           -- 全局：谓词触发后清空

窗口元数据函数：window_start()、window_end()。

事件时间（WITH 子句）

默认处理时间。指定事件时间字段与乱序/延迟容忍：

WITH (TIMESTAMP='event_time', TIMEUNIT='ms', MAXOUTOFORDERNESS='5s', ALLOWEDLATENESS='2s', IDLETIMEOUT='5s')

TIMEUNIT：ns/ms(默认)/ss/mi/hh/dd。

聚合与内置函数

聚合：COUNT(*) SUM AVG MAX MIN STDDEV STDDEVS VAR VARS MEDIAN PERCENTILE(f,p) COLLECT FIRST_VALUE LAST_VALUE MERGE_AGG DEDUPLICATE(f,bool)。

内置：数学 ABS/ROUND/FLOOR/CEIL/SQRT/POWER、字符串 CONCAT/UPPER/LOWER/LENGTH/SUBSTRING/TRIM、CAST(expr AS STRING)。

元数据表与流-表 JOIN

节点配置 tables 注册元数据表，SQL 里 JOIN 富化流行（transform 逐行富化、aggregator 富化后聚合）。streamsql 1.0.0+。

"tables": [
  {"name": "meta", "source": "file", "path": "/etc/rulego/device_meta.json", "format": "json", "refresh": "30s"}
]

JOIN 仅支持等值 ON（复合键用 AND），表侧列按别名命名空间返回（m.location）。刷新失败保留旧快照。

SELECT deviceId, m.location FROM stream s LEFT JOIN meta m ON s.deviceId = m.deviceId WHERE temperature > 30

规则链示例

每 30 秒按设备分组计算平均温度，超阈值告警：

{
  "ruleChain": {"id": "temp_aggregation", "name": "温度聚合监控", "root": true},
  "metadata": {
    "nodes": [
      {"id": "node_1", "type": "x/streamAggregator", "name": "温度聚合",
       "configuration": {"sql": "SELECT deviceId, AVG(temperature) as avg_temp, MAX(temperature) as max_temp, COUNT(*) as count FROM stream GROUP BY deviceId, TumblingWindow('30s')"}},
      {"id": "node_2", "type": "jsTransform", "name": "告警判断",
       "configuration": {"jsScript": "var r = JSON.parse(msg.data || msg); if (Array.isArray(r)) { r.forEach(function(x){ x.alert = x.max_temp > 35; }); msg = r; } return {'msg':msg,'metadata':metadata,'msgType':msgType};"}},
      {"id": "node_3", "type": "log", "name": "记录", "configuration": {"jsScript": "return JSON.stringify(msg);"}}
    ],
    "connections": [
      {"fromId": "node_1", "toId": "node_2", "type": "stream_event"},
      {"fromId": "node_2", "toId": "node_3", "type": "Success"}
    ]
  }
}

注意事项（高频踩坑）

• 输入必须是 JSON 类型（msg.DataType == JSON）。非 JSON 静默走 Failure、无日志、整条链无输出——极易踩坑。
  • jsTransform 造数据喂 stream 节点时，返回必须带 dataType:'JSON'（否则继承上游 DataType，上游若是 TEXT 就静默失败）：

    return {'msg': data, 'metadata': metadata, 'msgType': msgType, 'dataType': 'JSON'};
    

• SQL 字段直接用字段名（temperature），不要 msg. 前缀（与 jsFilter/jsTransform 的 msg.temperature 不同）。
• 字段名区分大小写，与 JSON key 精确匹配。
• FROM 后的名称是逻辑概念（通常用 stream）。
• 时间窗口参数：'30s'/'5m'/'1h'/'1d'。
• 不支持：流-流 JOIN（WITHIN 区间）、UNION、子查询、INSERT/UPDATE/DELETE、CREATE TABLE。