apache / apache/iceberg

Spark: MERGE INTO Statements with only WHEN NOT MATCHED Clauses are always executed at Snapshot Isolation

未关闭
#12,653 17 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
bug spark
主要语言
Java
星标
9.2k
派生
3.5k
平均合并
2 天 11 小时
30 天内合并 PR
132

描述

### Apache Iceberg version

1.8.1 (latest release)

### Query engine

Spark

### Please describe the bug 🐞

When running two concurrent `MERGE INTO` operations on an Apache Iceberg table, I expect them to be **idempotent** -- meaning Iceberg should either detect conflicts and resolve them or fail one of the jobs to prevent data inconsistencies.

However, Iceberg determines the **operation type dynamically** based on the result of the join condition, which can lead to unexpected behavior:
- If a match is found, Iceberg treats it as an **overwrite** operation and fails the second job due to conflicting commits.
- If no match is found, Iceberg considers it an **append** operation and attempts to resolve conflicts by creating a new manifest for appended data, as explained in the [Cost of Retries doc](https://iceberg.apache.org/docs/latest/reliability/#cost-of-retries).

This behavior introduces a problem:
If the dataset is large enough and neither job finds a match, both will proceed with appending data independently, causing **duplicate records**.

#### **Reproduction Steps**
Running the following query in concurrent jobs can result in duplicate data if no matching records exist in `dest`:
```SQL
MERGE INTO dest
USING src
ON dest.id = src.id
WHEN NOT MATCHED THEN
INSERT *
-- even with update action, we'll have the same issue
-- WHEN MATCHED THEN
-- UPDATE SET *
```
I initially expected the **operation type** to be determined by the query itself (i.e., always "append" in the query without `UPDATE` action). However, through testing, I found that Iceberg decides the operation type **at runtime**, based on the actual join results. This makes `MERGE INTO` **non-idempotent**, leading to unintended duplicate inserts.

#### **Expected Behavior**
Iceberg should ensure idempotency for `MERGE INTO`, preventing duplicate data when no matches are found.

#### **Additional Context**
- Iceberg version: 1.8.1
- Iceberg catalog: Glue catalog (type `glue`) with S3 FileIO
- Spark version: 3.5.5

Would love to hear if others have encountered this or if there's a recommended workaround.

### Willingness to contribute

- [ ] I can contribute a fix for this bug independently
- [x] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time

贡献指南

打开贡献指南

调研方向

从 issue 中描述的针对 Iceberg 表的并发 Spark MERGE INTO 复现开始,使用仅包含 WHEN NOT MATCHED 的查询以及指定的 Spark 和 Iceberg 版本。阅读 Cost of Retries 文档,并跟踪 runtime join 结果如何选择 append 或 overwrite 行为;当并发 no-match merge 不会创建重复记录,且该行为已由回归测试覆盖时,即表示完成。

由索引模型根据 Issue 内容生成。

评估

技术栈
java, spark, sql
领域
data-engineering, databases, distributed-systems
Issue 类型
缺陷
难度
4/5
预计耗时
3-5 天
活跃度
冷清
描述清晰度
基本清楚
新手友好度
48/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。