createStatusEngine

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

Creates a status helper engine that maintains the latest record for each key defined by keyColumn. The engine retains only the most recent record for each key, representing its latest state. Regardless of the outputPattern setting, the engine always maintains an internal latest state table that retains only the last record for each key. You can query the current latest state of all keys via select * from engine or select * from getStreamEngine(name).

The engine does not parse any status flags in the input. Even if the input table contains a column named status, it is treated as a regular column. The status is determined automatically based on whether the key already exists in the engine.

createStatusEngine itself does not perform any aggregation. It only marks the data stream with NEW/UPDATE or deduplicates records by key. In contrast, the statusColumn parameter of createTimeSeriesEngine / createReactiveStateEngine allows a streaming engine to read and process NEW/UPDATE flags in the input, thereby changing the computation behavior. For example, when an UPDATE record arrives, the engine replaces the latest value in the current window with the new value and recalculates the result, instead of accumulating the record as a new value.

A typical collaboration pattern is as follows: if the upstream data source does not carry status flags, use createStatusEngine first to automatically add NEW/UPDATE flags, and then pass its output to a downstream time series engine/reactive state engine to process the status semantics and perform correct aggregation.

Scenarios where createStatusEngine is applicable:

• The downstream consumer is not a time series engine or a reactive state engine, such as a user-defined handler, a keyed table, or a third-party consumer, and requires a generic status-labeling source.

• Records need to be deduplicated when the key changes. That is, when records with the same key arrive consecutively, only the last record is kept and output when the key changes.

• Multiple downstream engines need to share the same status labels, avoiding repeated checks in each engine.

• Only NEW/UPDATE labeling is required, and no aggregation is needed.

Output Modes

The status helper engine supports two output modes, specified by the outputPattern parameter:

• "all" (default): each input record produces one output record. The engine automatically appends a SYMBOL status column to the end of the output:

  • If the key appears for the first time, NEW is output.

  • If the key already exists, UPDATE is output.

  Therefore, the schema of outputTable must be all input columns plus an additional SYMBOL column at the end.

• "keyChange": only when the key changes, the last record of the previous key segment is output. No status column is output. Consecutive input records with the same key only overwrite the internal cache and do not trigger output. The schema of outputTable is the same as that of the input table, and no additional column is required.

name is a STRING scalar specifying the name of the engine. It is the only identifier of an engine on a data or compute node. It can contain letters, numbers, and "_" and must start with a letter.

dummyTable is a table object whose schema must be the same as the input stream table. Whether dummyTable contains data does not matter.

outputTable is a table to which the engine inserts calculation results. It can be an in-memory table or a DFS table. Create an empty table and specify the column names and types before calling the function.

The output columns are in the following order:

• Input columns. They are the same as all columns in dummyTable.

• Status column. If outputPattern="all", specify a SYMBOL column at the end. The engine automatically writes NEW or UPDATE to this column. If outputPattern="keyChange", this column is not required.

keyColumn is a STRING scalar or vector specifying the grouping column name(s). The engine maintains the latest state for each key based on this column and determines whether an input record is NEW or UPDATE.

outputPattern (optional) is a STRING scalar with the default value "all". It specifies the output mode of the engine. Supported values are:

• "all": each input record outputs one result, with a status column automatically generated by the engine.

• "keyChange": only when the key changes, the last record of the previous key segment is output. No status column is output.

A table object. Data is injected into the status helper engine by writing to this table object.

Example 1. This example demonstrates the behavior of createStatusEngine in the default outputPattern="all" mode. It simulates a stock quote stream. The engine determines whether each record is the first occurrence (NEW) or an update of an existing key (UPDATE) based on the sym column, and automatically appends a status column to the end of the output table.

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

// Define the input table schema
dummy = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])
// Define the output table: one more SYMBOL status column than the input table
output = table(10:0, `sym`time`price`status, [SYMBOL, TIMESTAMP, DOUBLE, SYMBOL])

// Create a status helper engine grouped by sym
engine = createStatusEngine(
    name="statusDemo",
    dummyTable=dummy,
    outputTable=output,
    keyColumn=`sym
)

// Construct sample data: A appears three times and B appears once
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

• Record 1: A appears for the first time, so NEW is output.

• Record 2: A already exists, so UPDATE is output.

• Record 3: B appears for the first time, so NEW is output.

• Record 4: A already exists, so UPDATE is output, regardless of whether other keys have appeared in between.

At this point, query the latest internal state of the engine. Only the last record is kept for each key:

select * from engine

Example 2. Input contains a business column named status. This example demonstrates that when the input table itself contains a business column named status, createStatusEngine does not parse this column or treat it as a NEW/UPDATE status flag. The column is output as a regular business column. The status generated by the engine must be written to an additional SYMBOL column at the end of outputTable, such as rowStatus.

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

// Define the input table schema: status is a regular business column, not a status column generated by the engine
dummy = table(10:0, `sym`time`status`price, [SYMBOL, TIMESTAMP, STRING, DOUBLE])
// Define the output table: keep the business status column and add rowStatus at the end for engine-generated NEW/UPDATE
output = table(10:0, `sym`time`status`price`rowStatus, [SYMBOL, TIMESTAMP, STRING, DOUBLE, SYMBOL])

// Create a status helper engine grouped by sym
engine = createStatusEngine(
    name="statusBusinessDemo",
    dummyTable=dummy,
    outputTable=output,
    keyColumn=`sym
)

// Construct sample data: raw1/raw2/raw3 in the status column are business values and are preserved as-is
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

As shown above, the values raw1, raw2, and raw3 in the status column are preserved as-is. rowStatus is the status column generated by the engine based on whether the key already exists. The first A record is marked as NEW, the second A record is marked as UPDATE, and the first B record is marked as NEW.

Example 3. This example demonstrates the outputPattern="keyChange" mode. This mode applies when input records arrive in consecutive segments by the same key. When the key changes, the engine outputs the last record of the previous key segment, achieving ordered deduplication.

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

dummy = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])
// In keyChange mode, no status column is output, and outputTable has the same schema as the input
output = table(10:0, `sym`time`price, [SYMBOL, TIMESTAMP, DOUBLE])

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

// Sample data: three consecutive A records, two consecutive B records, and one C record
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

• The A segment contains three consecutive records. When the key changes from A to B, the engine outputs the last record of the A segment (price=10.3).

• The B segment contains two consecutive records. When the key changes from B to C, the engine outputs the last record of the B segment (price=20.2).

• The C segment contains only one record. Because no subsequent data arrives to trigger a key change, the C segment is not output yet.

Related Functions: createTimeSeriesEngine, createReactiveStateEngine