如何为DolphinDB响应式状态引擎结果表添加自增ID列？

在DolphinDB的Reactive State Engine进行实时数据处理时，需要为每只股票维护一个自增ID字段作为标识，但createReactiveStateEngine函数无相关参数支持向结果表添加自增字段。
起初考虑使用cumcount函数统计累计记录，但因因子计算逻辑会过滤部分记录，该方法无法生效。
随后尝试了以下函数实现：

def setid(bd)
{   
    n = size(bd)
    result = array(INT, n)
    for(i in 0..(n-1)){
        id= select count(bondcode) as id from shared_tab_bond where bondcode=bd[i]   
        result[i]=id.id[i]
    }
    return result
}

注：shared_tab_bond是结果表，由于它是共享表，可直接在函数中引用。

但上述函数返回结果不正确，请问如何正确实现该功能？

原函数问题分析 #

• 并发安全问题：shared_tab_bond是实时写入的共享表，多线程并发写入时，循环内的count查询会读取到旧数据，导致ID重复或跳号。
• 性能瓶颈：循环逐条查询共享表会产生大量IO操作，严重降低实时处理的吞吐量。

正确实现方式 #

方式1：原子变量维护单股计数器 #

为每只股票独立维护原子计数器，利用atomicAdd保证并发场景下的计数准确性：

// 初始化全局共享字典，存储各股票的原子计数器
share dict(STRING, ATOMICINT) as bondIdCounter

def generateBondId(bd){
    n = size(bd)
    result = array(INT, n)
    for(i in 0..n-1){
        code = bd[i]
        // 首次处理该股票时，初始化计数器为1
        if(!bondIdCounter.contains(code)){
            bondIdCounter[code] = atomic(1L)
        }
        // 原子自增并返回当前ID值
        result[i] = atomicAdd(bondIdCounter[code], 1)
    }
    return result
}

• 优势：并发安全，性能远高于查询共享表计数，适合高吞吐量实时场景。
• 注意事项：系统重启后，需从shared_tab_bond中读取每只股票的最大ID，重新初始化bondIdCounter，避免ID重置。

方式2：Reactive State Engine内置状态维护计数 #

在引擎的计算函数中直接维护分组（股票）的计数状态，无需依赖外部表：

// 定义状态更新函数，按bondcode分组维护自增ID
def updateState(state, data){
    // state为字典结构：键是bondcode，值是该股票当前的最大ID
    codes = data.bondcode
    ids = array(INT, size(codes))
    for(i in 0..size(codes)-1){
        code = codes[i]
        if(!state.contains(code)){
            state[code] = 0
        }
        state[code] += 1
        ids[i] = state[code]
    }
    // 将生成的ID与原数据合并后返回
    return select *, ids as bondId from data
}

// 创建Reactive State Engine，指定分组键为bondcode，初始状态为空字典
rse = createReactiveStateEngine(
    name="bondIdEngine",
    metrics=updateState,
    dummyTable=inputTable,
    keyColumn="bondcode",
    outputTable=shared_tab_bond,
    state=dict()
)

• 优势：状态与引擎绑定，无需额外维护全局变量，重启引擎时可通过restoreState恢复状态，保证数据一致性。
• 适用场景：所有计算逻辑都在Reactive State Engine内完成的实时处理场景。

内容的提问来源于stack exchange，提问作者Polly