createWindowJoinEngine

createWindowJoinEngine(name, leftTable, rightTable, outputTable, window, metrics, matchingColumn, [timeColumn], [useSystemTime=false], [garbageSize = 5000], [maxDelayedTime], [nullFill], [outputElapsedMicroseconds=false], [sortByTime], [closed], [snapshotDir], [snapshotIntervalInMsgCount], [cachedTableCapacity=1024], [keyPurgeFreqInSec], [timeoutTrigger])

创建流数据 window join 引擎。返回一个左、右两表实时 window join 后的表对象。

该引擎与 window join 存在以下区别：

• window join 仅支持聚合计算，window join 引擎支持聚合计算，也支持非聚合计算。
• 计算指标中的列未指定表名时，window join 默认取右表列，而 window join 引擎默认取左表列。

注入 window join 引擎左、右两表的数据，根据 matchingColumn 分组。在各分组内，左表的每条记录，都会同右表中一个时间区间（即窗口）内的数据进行连接，并根据指定的 metrics 计算窗口内的数据并输出。

• 普通窗口（window 不为 0:0）：

  右表的计算窗口将由左表当前数据的时间戳和 window 确定。假设左表当前记录的时间戳为 t，window 为 a:b，则右表时间戳属于 [t+a, t+b] 的数据将与左表当前记录连接并计算输出。

  触发计算的规则：

  • useSystemTime=false：
    • 同组数据触发：各分组当前窗口数据的计算将由该窗口结束后的第一条属于该分组的数据触发。触发计算的数据不参与该窗口的计算。
    • 其它分组数据触发：对于某个分组中未发生计算的窗口，若其窗口右边界 + maxDelayedTime < 右表最新收到的任意一个分组数据的时间戳，则该窗口的计算将被新收到的这条数据触发。
  • useSystemTime=true：当系统时间到达各分组未发生计算的窗口的右边界时，触发该窗口的计算。

• 特殊窗口（window 为 0:0）：

  右表的计算窗口将由左表当前数据和其上一条数据的时间戳决定。默认情况下，该窗口左闭右开，假设左表当前记录的时间戳为 t，上一条记录的时间戳为 t0，则右表计算窗口为 [t0, t)。可以通过指定参数 closed = “right”，设置窗口为左开右闭。 触发计算的规则：

  • useSystemTime=false：各分组当前窗口数据的计算将由窗口结束后的第一条属于该分组的右表数据触发。
  • useSystemTime=true：各个分组收到的左表数据将触发对应分组窗口的数据计算输出。

更多流数据引擎的应用场景说明可以参考 流计算引擎。

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

leftTable 表对象。可以不包含数据，但结构必须与输入的流数据表相同。2.00.9.3 版本开始支持 array vector 类型。

rightTable 表对象。可以不包含数据，但结构必须与输入的流数据表相同。

outputTable 必选参数，为计算结果的输出表。window join 引擎会将计算结果插入该表。

• 输出表各列的顺序如下：
  1. 时间列。其中：
     • 若 useSystemTime = true，为 TIMESTAMP 类型；
     • 若 useSystemTime = false，数据类型与 timeColumn 列一致。
  2. 连接列。与 matchingColumn 中的列以及其顺序一致，可为多列。
  3. 计算结果列。可为多列。
  4. 耗时列。若指定 outputElapsedMicroseconds = true，则需要增加一个 LONG 类型和一个 INT 类型的列，分别用于存储引擎内部每个 batch 的数据耗时（单位：微秒）和记录数。

window 必选参数，表示滑动窗口区间的整型或 DURATION 数据对，其中左右边界都包含在内。

timeColumn 可选参数，当 useSystemTime = false 时，指定要连接的两个表中时间列的名称。leftTable 和 rightTable 时间列名称可以不同，但数据类型需保持一致。当 leftTable 和 rightTable 时间列名称不同时，timeColumn 为一个长度为2的字符串向量。

useSystemTime 可选参数，表示 outputTable 中第一列（时间列）为系统当前时间（useSystemTime = true）或左表的时间列（useSystemTime = false）。

garbageSize 可选参数，是正整数，默认值是5,000（单位为行）。随着订阅的流数据不断积累进入 window join 引擎，存放在内存中的数据会越来越多，这时需要清理不再需要的历史数据。当左/右两表各个分组内的数据行数超过 garbageSize 值时，系统会清理本次计算不需要的历史数据。

maxDelayedTime 可选参数，是正整数，用于触发引擎中长时间未输出的分组数据进行计算。 具体来说，若(某个分组中未发生计算的窗口右边界) + (maxDelayedTime) < (右表最新收到的任意一个分组数据的时间戳)，则这条数据会触发该窗口计算输出。

指定该参数时，必须同时指定 timeColumn，且两者的单位需一致。默认值为3秒，根据 timeColumn 的精度换算。例如，若 timeColumn 的精度是毫秒，则默认值为3000毫秒。

nullFill 和输出表列字段等长且类型一一对应的元组，用于填充以下列中的空值：输出表中包含的左表列、右表列、右表列被聚合计算后的计算结果列。

outputElapsedMicroseconds 布尔值，表示是否输出每个 batch 中数据从注入引擎到计算输出的总耗时，以及每个 batch 包含的总记录数，默认为 false。指定参数 outputElapsedMicroseconds = true 后，在定义 outputTables 时需要在最后增加两列，详见 outputTable 参数说明。

sortByTime 布尔值，表示是否按全局时间顺序输出数据。默认值为 false，表示不按全局时间输出数据，仅在组内按时间顺序输出数据。注意：当设置 sortByTime=true 时，必须保证输入的左表和右表的数据全局有序，且不可设置 maxDelayedTime。

若要开启快照机制（snapshot），必须指定 snapshotDir 与 snapshotIntervalInMsgCount。

snapshotDir 可选参数，字符串，表示保存引擎快照的文件目录。

• 指定的目录必须存在，否则系统会提示异常。
• 创建流数据引擎时，如果指定了 snapshotDir，会检查该目录下是否存在快照。如果存在，会加载该快照，恢复引擎的状态。
• 多个引擎可以指定同一个目录存储快照，用引擎的名称来区分快照文件。
• 一个引擎的快照可能会使用三个文件名：
  • 临时存储快照信息：文件名为 <engineName>.tmp；
  • 快照生成并刷到磁盘：文件保存为 <engineName>.snapshot；
  • 存在同名快照：旧快照自动重命名为 <engineName>.old。

snapshotIntervalInMsgCount 可选参数，为整数类型，表示每隔多少条数据保存一次流数据引擎快照。

cachedTableCapacity 可选参数，为正整数，表示引擎为每个不同的分组分别创建的左缓存表和右缓存表的初始容量（以数据条数计）。默认值为 1024。

timeoutTrigger 可选参数，为布尔标量，表示当右表在较长时间内无数据输入时，是否主动触发计算。开启后可确保所有数据均能在其窗口的右边界 + [1,3) * maxDelayedTime 时间范围内触发计算。启用timeoutTrigger 时，需满足以下条件：

• useSystemTime=false

• sortByTime=false

• maxDelayedTime 为正整数。

返回一个表对象。

例1. 本例以股票代码（sym）为键，针对左表中的每一笔最新成交价格，在右表中寻找与该成交时间点偏差在±2毫秒内的所有数值数据，并计算这一微小时间窗口内的数值总和。通常用于衡量即时流动性或短期冲击成本，例如当一笔大单成交时，观察紧接着的几毫秒内市场的反应深度。

share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

nullFill= [2012.01.01T00:00:00.000, `NONE, 0.0, 0.0, 0.0]
wjEngine=createWindowJoinEngine(name="test1", leftTable=leftTable, rightTable=rightTable, outputTable=output,  window=-2:2, metrics=<[price,val,sum(val)]>, matchingColumn=`sym, timeColumn=`time, useSystemTime=false,nullFill=nullFill)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

n=10
tp1=table(take(2012.01.01T00:00:00.000+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(NULL join rand(10.0, n-1),2*n) as price)
tp1.sortBy!(`time)
leftTable.append!(tp1)

tp2=table(take(2012.01.01T00:00:00.000+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(double(1..n),2*n) as val)
tp2.sortBy!(`time)
rightTable.append!(tp2)

select * from output where time between 2012.01.01T00:00:00.000:2012.01.01T00:00:00.001

例2. 下例展示特殊窗口（window=0:0）的计算。引擎仅对左右两张表中时间戳完全一致的记录进行关联。常用于数据清洗，例如将来自不同供应商在同一纳秒时刻的快照进行合并。

share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable

v = [1, 5, 10, 15]
tp1=table(2012.01.01T00:00:00.000+v as time, take(`A   , 4) as sym, rand(10.0,4) as price)

v = [1, 2, 3, 4, 5, 6, 9, 15]
tp2=table(2012.01.01T00:00:00.000+v as time, take(`A   , 8) as sym, rand(10.0,8) as val)

share table(100:0, `time`sym`price`val`sum_val, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE[], DOUBLE]) as output
wjEngine=createWindowJoinEngine(name="test1", leftTable=leftTable, rightTable=rightTable, outputTable=output,  window=0:0, metrics=<[price, val, sum(val)]>, matchingColumn=`sym, timeColumn=`time, useSystemTime=false)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

leftTable.append!(tp1)
rightTable.append!(tp2)

例3. 例 2 中展示了 window=0:0 时的默认情况，该窗口左闭右开。与上例类似，但通过参数 closed="right" 将匹配规则改为左开右闭。在数据时间戳发生跳变时，决定了某一时刻的记录是归属于上一个窗口还是下一个窗口。

unsubscribeTable(tableName="leftTable", actionName="joinLeft")
unsubscribeTable(tableName="rightTable", actionName="joinRight")
undef(`leftTable,SHARED)
undef(`rightTable,SHARED)
dropAggregator(name="test1")

share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable

v1 = [1, 5, 10, 15]
tp1=table(2012.01.01T00:00:00.000+v1 as time, take(`A, 4) as sym, rand(10.0,4) as price)

v2 = [1, 2, 3, 4, 5, 6, 9, 15]
tp2=table(2012.01.01T00:00:00.000+v2 as time, take(`A, 8) as sym, rand(10.0,8) as val)

share table(100:0, `time`sym`price`val`sum_val, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE[], DOUBLE]) as output
wjEngine=createWindowJoinEngine(name="test1", leftTable=leftTable, rightTable=rightTable, outputTable=output,  window=0:0, metrics=<[price, val, sum(val)]>, matchingColumn="sym", timeColumn="time", useSystemTime=false, closed="right")

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

leftTable.append!(tp1)
rightTable.append!(tp2)
sleep(100)
select * from output

返回：

time	                  sym	price	val	                      sum_val
2012.01.01T00:00:00.001	A	9.7366	[7.8310]	                  7.831
2012.01.01T00:00:00.005	A	2.6537	[1.8564,4.6238,8.2536,3.1028]     17.8368
2012.01.01T00:00:00.010	A	3.9586	[0.8413,8.0684]	           8.9098

例4. 下例展示指定 sortByTime = true 时，引擎将按时间顺序输出数据，解决了流式计算中结果乱序的问题。

//清理引擎及变量
unsubscribeTable(tableName="leftTable", actionName="joinLeft")
unsubscribeTable(tableName="rightTable", actionName="joinRight")
undef(`leftTable,SHARED)
undef(`rightTable,SHARED)
dropAggregator(name="test1")

//定义引擎
share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output
nullFill= [2012.01.01T00:00:00.000, `NONE, 0.0, 0.0, 0.0]
wjEngine=createWindowJoinEngine(name="test1", leftTable=leftTable, rightTable=rightTable, outputTable=output,  window=-2:2, metrics=<[price,val,sum(val)]>, matchingColumn=`sym, timeColumn=`time, useSystemTime=false,nullFill=nullFill, sortByTime=true)

//定义数据
subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

n=10
tp1=table(take(2012.01.01T00:00:00.000+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(NULL join rand(10.0, n-1),2*n) as price)
tp1.sortBy!(`time)
leftTable.append!(tp1)

tp2=table(take(2012.01.01T00:00:00.000+0..10, 2*n) as time, take(`A, n) join take(`B, n) as sym, take(double(1..n),2*n) as val)
tp2.sortBy!(`time)
rightTable.append!(tp2)

sleep(100)
select * from output where time between 2012.01.01T00:00:00.000:2012.01.01T00:00:00.001

例5. 本例展示 matchingColumn为多个连接列的情况，模拟了跨市场或多维度的数据融合场景。基于例 1，本例设置 matchingColumn=[`exchange, `sym]，同时使用 exchange 和 sym 两列作为连接条件。

// 创建左表
share streamTable(1:0, `time`exchange`sym`price, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as leftTable
// 创建右表
share streamTable(1:0, `time`exchange`sym`val, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as rightTable
// 输出表
share table(100:0, `time`exchange`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

nullFill = [2012.01.01T00:00:00.000, `NONE, `NONE, 0.0, 0.0, 0.0]

// 使用2个连接列：exchange 和 sym
wjEngine = createWindowJoinEngine(
    name="test1", 
    leftTable=leftTable, 
    rightTable=rightTable, 
    outputTable=output,  
    window=-2:2, 
    metrics=<[price, val, sum(val)]>, 
    matchingColumn=[`exchange, `sym],  // 多个连接列
    timeColumn=`time, 
    useSystemTime=false,
    nullFill=nullFill
)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

n=10
// 左表数据
tp1=table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`SSE`SZSE, n) join take(`SSE`SZSE, n) as exchange,
    take(`A, n) join take(`B, n) as sym, 
    take(NULL join rand(10.0, n-1), 2*n) as price
)
tp1.sortBy!(`time)
leftTable.append!(tp1)

// 右表数据
tp2=table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`SSE`SZSE, n) join take(`SSE`SZSE, n) as exchange,
    take(`A, n) join take(`B, n) as sym, 
    take(double(1..n), 2*n) as val
)
tp2.sortBy!(`time)
rightTable.append!(tp2)

select * from output where time between 2012.01.01T00:00:00.000:2012.01.01T00:00:00.001

matchingColumn 也支持左右表连接列名不同时的情况。下面设置 matchingColumn=[[exchange, sym], [market, symbol]]，左表使用 exchange 和 sym 作为连接列，而右表使用 market 和 symbol 作为连接列，输出表列名为 exchange 和 sym。

// 创建左表
share streamTable(1:0, `time`exchange`sym`price, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as leftTable
// 创建右表
share streamTable(1:0, `time`market`symbol`val, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE]) as rightTable
// 输出表
share table(100:0, `time`exchange`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

nullFill = [2012.01.01T00:00:00.000, `NONE, `NONE, 0.0, 0.0, 0.0]

// 左右表列名不同时，matchingColumn 使用 tuple 形式指定
wjEngine = createWindowJoinEngine(
    name="test2", 
    leftTable=leftTable, 
    rightTable=rightTable, 
    outputTable=output,  
    window=-2:2, 
    metrics=<[price, val, sum(val)]>, 
    matchingColumn=[[`exchange, `sym], [`market, `symbol]],  // 左表列名对应右表列名
    timeColumn=`time, 
    useSystemTime=false,
    nullFill=nullFill
)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

n=10
// 左表数据
tp1=table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`SSE`SZSE, n) join take(`SSE`SZSE, n) as exchange,
    take(`A, n) join take(`B, n) as sym, 
    take(NULL join rand(10.0, n-1), 2*n) as price
)
tp1.sortBy!(`time)
leftTable.append!(tp1)

// 右表数据
tp2=table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`SSE`SZSE, n) join take(`SSE`SZSE, n) as market,  // 对应左表的 exchange
    take(`A, n) join take(`B, n) as symbol,  // 对应左表的 sym
    take(double(1..n), 2*n) as val
)
tp2.sortBy!(`time)
rightTable.append!(tp2)

// 查看输出表
select * from output where time between 2012.01.01T00:00:00.000:2012.01.01T00:00:00.001

例6. 本例设置 maxDelayedTime=5（5 毫秒），如果某个股票的窗口结束后超过 5 毫秒还未收到该股票的新数据，但其他股票有新数据到达时，该窗口会被强制触发计算并输出结果，保证了实时系统的低延迟特性。

// 创建表结构
share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

nullFill = [2012.01.01T00:00:00.000, `NONE, 0.0, 0.0, 0.0]

// 设置 maxDelayedTime=5 毫秒
wjEngine = createWindowJoinEngine(
    name="test3", 
    leftTable=leftTable, 
    rightTable=rightTable, 
    outputTable=output,  
    window=-2:2, 
    metrics=<[price, val, sum(val)]>, 
    matchingColumn=`sym, 
    timeColumn=`time, 
    useSystemTime=false,
    nullFill=nullFill,
    maxDelayedTime=5  // 设置 maxDelayedTime=5 毫秒
)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

// 左表数据
tp1_A = table(
    2012.01.01T00:00:00.000 + [0, 1, 2, 3, 4, 5] as time,
    take(`A, 6) as sym,
    rand(10.0, 6) as price
)
tp1_B = table(
    2012.01.01T00:00:00.000 + [0, 20] as time,
    take(`B, 2) as sym,
    rand(10.0, 2) as price
)
tp1 = tp1_A.unionAll(tp1_B)
tp1.sortBy!(`time)
leftTable.append!(tp1)

// 右表数据
tp2_A = table(
    2012.01.01T00:00:00.000 + [0, 1, 2, 3, 4, 5, 6, 7] as time,
    take(`A, 8) as sym,
    double(1..8) as val
)
tp2_B = table(
    2012.01.01T00:00:00.000 + [0, 1, 20] as time,
    take(`B, 3) as sym,
    double([10, 11, 12]) as val
)
tp2 = tp2_A.unionAll(tp2_B)
tp2.sortBy!(`time)
rightTable.append!(tp2)

sleep(100)
select * from output order by time, sym

例7. 基于例 6，本例展示 timeoutTrigger 参数的作用。与 maxDelayedTime 不同的是，timeoutTrigger 会控制右表长时间无数据输入时主动触发计算，即使没有其他分组的新数据到达，延迟的窗口也能被强制计算。这非常适用于盘中盯市或实时风控，确保市场没有交易也能定时输出最新的资产估值和风险敞口。

// 创建表结构
share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable
share table(100:0, `time`sym`factor1`factor2`factor3, [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE]) as output

nullFill = [2012.01.01T00:00:00.000, `NONE, 0.0, 0.0, 0.0]

// 启用timeoutTrigger
wjEngine = createWindowJoinEngine(
    name="test3", 
    leftTable=leftTable, 
    rightTable=rightTable, 
    outputTable=output,  
    window=-2:2, 
    metrics=<[price, val, sum(val)]>, 
    matchingColumn=`sym, 
    timeColumn=`time, 
    useSystemTime=false,
    nullFill=nullFill,
    maxDelayedTime=5,
    timeoutTrigger=true  // 启用超时触发
)

subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, handler=appendForJoin{wjEngine, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, handler=appendForJoin{wjEngine, false}, msgAsTable=true)

// 左表数据
// 股票A：时间戳 0, 1, 2, 3, 4, 5
tp1_A = table(
    2012.01.01T00:00:00.000 + [0, 1, 2, 3, 4, 5] as time,
    take(`A, 6) as sym,
    rand(10.0, 6) as price
)

// 股票B：时间戳 0, 20（长时间间隔）
tp1_B = table(
    2012.01.01T00:00:00.000 + [0, 20] as time,
    take(`B, 2) as sym,
    rand(10.0, 2) as price
)

tp1 = tp1_A.unionAll(tp1_B)
tp1.sortBy!(`time)
leftTable.append!(tp1)

// 右表数据
tp2_A = table(
    2012.01.01T00:00:00.000 + [0, 1, 2, 3, 4, 5, 6, 7] as time,
    take(`A, 8) as sym,
    double(1..8) as val
)

tp2_B = table(
    2012.01.01T00:00:00.000 + [0, 1, 20] as time,
    take(`B, 3) as sym,
    double([10, 11, 12]) as val
)

tp2 = tp2_A.unionAll(tp2_B)
tp2.sortBy!(`time)
rightTable.append!(tp2)

sleep(100)
select * from output order by time, sym

由于设置了 timeoutTrigger=true，本例在右表长时间无数据输入时主动触发计算，股票 B 在第 20 秒的延迟窗口被强制计算输出，因此输出表结果比例 6 多了一行 time 为 2012.01.01 00:00:00.020 的数据。

例8. 本例展示通过设置 outputElapsedMicroseconds=true 来监控 window join 引擎的性能。与例 1 相比，本例输出表末尾增加了两列：elapsedUs 记录每个 batch 从注入引擎到计算输出的总耗时（单位：微秒），batchRows 记录每个 batch 包含的总记录数。不仅有助于评估当前参数配置对性能的影响，还能在发生故障时快速定位问题，保障高频交易系统稳定运行。

// 创建表
share streamTable(1:0, `time`sym`price, [TIMESTAMP, SYMBOL, DOUBLE]) as leftTable
share streamTable(1:0, `time`sym`val, [TIMESTAMP, SYMBOL, DOUBLE]) as rightTable

// 输出表：增加 elapsedUs 和 batchRows 两列
share table(100:0, `time`sym`factor1`factor2`factor3`elapsedUs`batchRows, 
    [TIMESTAMP, SYMBOL, DOUBLE, DOUBLE, DOUBLE, LONG, INT]) as output

nullFill = [2012.01.01T00:00:00.000, `NONE, 0.0, 0.0, 0.0, 0l, 0]

// 创建引擎，设置 outputElapsedMicroseconds=true
wjElapsedDemo = createWindowJoinEngine(
    name="wjElapsedDemo", 
    leftTable=leftTable, 
    rightTable=rightTable, 
    outputTable=output,  
    window=-2:2, 
    metrics=<[price, val, sum(val)]>, 
    matchingColumn=`sym, 
    timeColumn=`time, 
    useSystemTime=false,
    nullFill=nullFill,
    outputElapsedMicroseconds=true  // 输出计算耗时
)

// 订阅流数据表
subscribeTable(tableName="leftTable", actionName="joinLeft", offset=0, 
    handler=appendForJoin{wjElapsedDemo, true}, msgAsTable=true)
subscribeTable(tableName="rightTable", actionName="joinRight", offset=0, 
    handler=appendForJoin{wjElapsedDemo, false}, msgAsTable=true)

// 左表数据
n=10
tp1 = table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`A, n) join take(`B, n) as sym, 
    take(NULL join rand(10.0, n-1), 2*n) as price
)
tp1.sortBy!(`time)
leftTable.append!(tp1)

// 右表数据
tp2 = table(
    take(2012.01.01T00:00:00.000+0..10, 2*n) as time, 
    take(`A, n) join take(`B, n) as sym, 
    take(double(1..n), 2*n) as val
)
tp2.sortBy!(`time)
rightTable.append!(tp2)

sleep(100)
select * from output where time between 2012.01.01T00:00:00.000:2012.01.01T00:00:00.001