使用griddb_python时GridDB时间序列容器并发写入机制问询

背景 #

使用GridDB Cloud和griddb_python SDK构建高并发数据摄入管道，多个分布式Worker向TIME_SERIES容器写入同一device_id的传感器数据，容器Schema定义如下：

store.put_container("telemetry", [
    ("timestamp", griddb.Type.TIMESTAMP),
    ("device_id", griddb.Type.STRING),
    ("rpm", griddb.Type.INTEGER),
    ("vibration", griddb.Type.FLOAT)
], griddb.ContainerType.TIME_SERIES, True)

每个Worker通过put_row()插入数据，偶尔需要更新/修正已插入的行（如延迟摄入、错误恢复场景）。

遇到的问题 #

• 未找到GridDB处理同一主键行（相同timestamp+device_id）并发写入/更新的明确文档说明；
• TIME_SERIES容器的锁或乐观并发控制机制在文档中未清晰记载，与传统RDBMS差异较大。

核心疑问 #

GridDB在TIME_SERIES容器执行put_row()或update_row()时，是否支持行级锁或乐观并发控制？若两个Worker写入同一timestamp+device_id的行，是最新写入生效还是行为未定义？

已尝试操作 #

• 在Python中模拟同一行的重叠写入测试，结果不稳定：有时静默覆盖原有数据，有时抛出GSException；
• 查阅GridDB文档（提及事务支持），但针对Python SDK和TIME_SERIES容器的细节说明不足；
• 排查行键约束引发的写入冲突，行为表现不一致。

解答 #

1. 行级锁支持 #

GridDB对TIME_SERIES容器的put_row()和update_row()操作支持行级锁，锁的粒度为单一行，由对应数据分区的主节点负责管理：

• 执行put_row()插入新行时，无锁竞争则正常写入；若行已存在，put_row()会触发更新逻辑，此时会尝试获取行级锁；
• 多个Worker并发操作同一行时，只有第一个成功获取锁的操作能完成写入/更新，其他竞争操作会抛出GSException（错误码通常为0x40000019，即CONCURRENT_UPDATE_ERROR）。

2. 并发写入的行为逻辑 #

需先确保timestamp+device_id被设为复合主键（否则TIME_SERIES默认仅以timestamp为主键），此时并发写入的行为是确定性的：

• 若行不存在：所有并发插入操作中，只有一个会成功创建行，其余会因主键冲突抛出异常；
• 若行已存在：成功获取行级锁的操作会覆盖原有数据（最新写入生效），未获取锁的操作会抛出并发异常。

你测试中出现的“有时静默覆盖、有时抛异常”的现象，本质是并发时间窗口的差异：当一个操作已完成锁释放与数据提交，后续操作会直接覆盖；若两个操作在同一锁持有窗口内竞争，则未抢到锁的操作会抛出异常。

3. 乐观并发控制的实现 #

GridDB本身未内置乐观锁的版本校验机制，但可以通过事务+自定义版本列实现乐观并发控制：

1. 在容器Schema中添加version列（类型为INTEGER），初始值设为0；
2. 更新行时，先在事务中读取当前行的version值；
3. 执行更新操作时，添加条件version = 当前读取值，同时将version自增；
4. 若更新行数为0，说明该行已被其他操作修改，触发重试逻辑。

示例代码片段：

# 开启事务
tx = store.start_transaction()
try:
    container = tx.get_container("telemetry")
    # 读取当前行的version
    query = container.query("select version where timestamp = ? and device_id = ?")
    query.set_params([target_ts, target_device])
    rs = query.fetch()
    current_version = rs.next()[0]
    # 带条件更新
    update_count = container.update_row(
        {"timestamp": target_ts, "device_id": target_device, "rpm": new_rpm, "vibration": new_vibration, "version": current_version + 1},
        condition=f"version = {current_version}"
    )
    if update_count == 0:
        # 版本不匹配，触发重试
        tx.rollback()
        # 此处添加重试逻辑
    else:
        tx.commit()
except Exception as e:
    tx.rollback()
    raise e

4. 关键注意事项 #

• 确保TIME_SERIES容器的复合主键为timestamp+device_id：创建容器时需通过options参数指定rowkey，否则无法保证同一device_id同一timestamp的行唯一性；
  示例修改后的容器创建代码：

  options = griddb.ContainerInfoOptions()
  options.set_rowkey(["timestamp", "device_id"])
  store.put_container("telemetry", [
      ("timestamp", griddb.Type.TIMESTAMP),
      ("device_id", griddb.Type.STRING),
      ("rpm", griddb.Type.INTEGER),
      ("vibration", griddb.Type.FLOAT)
  ], griddb.ContainerType.TIME_SERIES, True, options)
  

• 高并发场景下，建议捕获CONCURRENT_UPDATE_ERROR异常并实现重试逻辑，避免数据丢失；
• 自动提交模式下（is_auto_commit=True），每个put_row()/update_row()都是独立事务，锁的持有时间极短，竞争概率相对较低，但仍需处理异常。

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