createStatusEngine

createStatusEngine(name, dummyTable, outputTable, keyColumn, [outputPattern="all"])

创建一个按 keyColumn 维护最新记录的状态辅助引擎。引擎内部为每个 key 只保留最新一条记录（即“最新状态”）。无论 outputPattern 取何值，引擎内部始终维护一份最新状态表，每个 key 仅保留最后一条记录。可通过 select * from engine 或 select * from getStreamEngine(name) 查询当前所有 key 的最新状态。

引擎不解析输入中的任何状态标志，即使输入表包含名为 status 的列，也仅作为普通列处理。状态完全由引擎根据 key 是否已存在自动判定。

createStatusEngine 本身不做任何聚合计算，只负责为数据流标注 NEW/UPDATE 或按 key 去重。而 createTimeSeriesEngine / createReactiveStateEngine 的 statusColumn 参数是让流计算引擎读取并处理输入中的 NEW/UPDATE 标志，从而改变计算行为（例如遇到 UPDATE 时用新值替换窗口内最新值后重新聚合，而非累加）。

两者的典型协作方式：当上游数据源本身不携带状态标志时，先通过 createStatusEngine 自动补上 NEW/UPDATE，再将输出传入下游的时序聚合引擎/响应式状态引擎处理状态语义，以进行正确的聚合计算。

适用 createStatusEngine 的场景：

• 下游不是时序聚合引擎/响应式状态引擎（如自定义 handler、keyedTable 或第三方消费者），需要通用的状态标注源。

• 需要按 key 切换去重（即输入按同一 key 连续到来时只保留最后一条，key 变化时才输出）。

• 多个下游引擎共享同一份状态标注，避免每个引擎重复判断。

• 纯标注场景：只需知道“新增还是更新”，不需要做任何聚合。

输出模式

状态辅助引擎支持两种输出模式，通过参数 outputPattern 指定：

• "all"（默认）：每条输入均产生一条输出。引擎自动在输出末尾附加一个 SYMBOL 类型的状态列：

  • key 首次出现：输出 NEW。

  • key 已存在：输出 UPDATE。

  因此，outputTable 的 schema 必须为：全部输入列 + 末尾一个额外的 SYMBOL 列。

• "keyChange"：仅在 key 发生切换时输出上一个 key 段的最后一条记录，不输出状态列。同一 key 的连续输入只覆盖内部缓存，不触发输出。outputTable 的 schema 与输入表一致，无需额外列。

name 字符串标量，表示状态引擎的名称，作为其在一个数据节点/计算节点上的唯一标识。可包含字母、数字和下划线，但必须以字母开头。

dummyTable 一个表对象，和输入的流数据表的 schema 一致，可以含有数据，亦可为空表。

outputTable 一个表对象，用于接收引擎的输出结果。创建引擎之前，需要将输出表预先设立为一个空表，并指定各列的列名以及数据类型。

输出表的列顺序如下：

• 输入列。与 dummyTable 的全部列保持一致。

• 状态列。若 outputPattern="all"，则在末尾指定一个 SYMBOL 类型的列，由引擎自动写入 NEW 或 UPDATE。若 outputPattern="keyChange"，则无需该列。

keyColumn 字符串标量或向量，表示分组列名。引擎根据该列维护每个 key 的最新状态，并判定输入记录是 NEW 还是 UPDATE。

outputPattern 可选参数，字符串标量，默认值为 "all"。指定引擎的输出模式，支持以下取值：

• "all"：每条输入均输出一条结果，附带引擎自动生成的状态列。

• "keyChange"：仅在 key 切换时输出上一个 key 段的最后一条记录，不输出状态列。

一个表对象，通过向该表对象写入，将数据注入状态辅助引擎。

例 1. 本例演示 createStatusEngine 在默认 outputPattern="all" 模式下的行为。模拟一个股票行情流，引擎根据 sym 列判断每条记录是首次出现（NEW）还是已有更新（UPDATE），并在输出表末尾自动附加状态列。

try{ dropStreamEngine(`statusDemo) } catch(ex){}

// 定义输入表结构
dummy = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])
// 定义输出表：比输入表多一个 SYMBOL 类型的状态列
output = table(10:0, `sym`time`price`status, [SYMBOL, TIMESTAMP, DOUBLE, SYMBOL])

// 创建状态辅助引擎，按 sym 分组
engine = createStatusEngine(
    name="statusDemo",
    dummyTable=dummy,
    outputTable=output,
    keyColumn=`sym
)

// 构造模拟数据：A 出现 3 次，B 出现 1 次
input = table(`A`A`B`A as sym,
              2026.07.01T09:30:01.000 + 0 1000 2000 3000 as time,
              10.1 10.2 20.1 10.3 as price)
engine.append!(input)

select * from output

• 第 1 条记录：A 首次出现，输出 NEW。

• 第 2 条记录：A 已存在，输出 UPDATE。

• 第 3 条记录：B 首次出现，输出 NEW。

• 第 4 条记录：A 已存在（无论中间是否有其他 key 插入），输出 UPDATE。

此时查询引擎内部的最新状态，每个 key 只保留最后一条记录：

select * from engine

例 2. 输入存在业务 status 列。本例演示当输入表本身包含名为 status 的业务列时，createStatusEngine 不会解析该列，也不会将其作为 NEW/UPDATE 状态标志处理。该列会作为普通列原样输出；引擎生成的状态需要通过 outputTable 末尾额外添加的 SYMBOL 列（如 rowStatus）输出。

try{ dropStreamEngine(`statusBusinessDemo) } catch(ex){}

// 定义输入表结构：status 是普通业务列，不是引擎生成的状态列
dummy = table(10:0, `sym`time`status`price, [SYMBOL, TIMESTAMP, STRING, DOUBLE])
// 定义输出表：保留业务 status 列，并在末尾额外添加 rowStatus 用于存储引擎生成的 NEW/UPDATE
output = table(10:0, `sym`time`status`price`rowStatus, [SYMBOL, TIMESTAMP, STRING, DOUBLE, SYMBOL])

// 创建状态辅助引擎，按 sym 分组
engine = createStatusEngine(
    name="statusBusinessDemo",
    dummyTable=dummy,
    outputTable=output,
    keyColumn=`sym
)

// 构造模拟数据：status 列中的 raw1/raw2/raw3 是业务值，会被原样保留
input = table(`A`A`B as sym,
              2026.07.01T09:30:01.000 + 0 1000 2000 as time,
              "raw1" "raw2" "raw3" as status,
              10.1 10.2 20.1 as price)
engine.append!(input)

select * from output

可以看到，status 列中的 raw1、raw2、raw3 被原样保留；rowStatus 才是引擎根据 key 是否已存在自动生成的状态列。A 第一次出现时输出 NEW，第二次出现时输出 UPDATE；B 第一次出现时输出 NEW。

例 3. 本例演示 outputPattern="keyChange" 模式。该模式适用于输入数据按同一 key 连续成段到来的场景。引擎在 key 发生切换时，输出上一个 key 段的最后一条记录，实现有序去重。

try{ dropStreamEngine(`statusDedupDemo) } catch(ex){}

dummy = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])
// keyChange 模式不输出状态列，outputTable schema 与输入一致
output = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])

engine = createStatusEngine(
    name="statusDedupDemo",
    dummyTable=dummy,
    outputTable=output,
    keyColumn=`sym,
    outputPattern="keyChange"
)

// 模拟数据：A 连续 3 条，B 连续 2 条，C 1 条
input = table(`A`A`A`B`B`C as sym,
              2026.07.01T09:30:01.000 + 0 1000 2000 3000 4000 5000 as time,
              10.1 10.2 10.3 20.1 20.2 30.1 as price)
engine.append!(input)

select * from output

• A 段连续 3 条。当 key 从 A 切换到 B 时，输出 A 段的最后一条（price=10.3）。

• B 段连续 2 条。当 key 从 B 切换到 C 时，输出 B 段的最后一条（price=20.2）。

• C 段只有 1 条，由于之后没有新数据到来触发 key 切换，C 段暂未输出。

相关函数：createTimeSeriesEngine，createReactiveStateEngine