Spark: MERGE INTO Statements with only WHEN NOT MATCHED Clauses are always executed at Snapshot Isolation
- Vorherrschende Sprache
- Java
- Sterne
- 9.2k
- Forks
- 3.5k
- Ø Merge
- 2 T. 11 Std.
- Gemergte PRs (30 T.)
- 132
Beschreibung
### 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
Beitragsleitfaden
Rechercherichtung
Beginne mit der im Issue beschriebenen Reproduktion von gleichzeitig ausgeführten Spark MERGE INTO-Vorgängen gegen eine Iceberg-Tabelle, wobei die Abfrage verwendet wird, die nur WHEN NOT MATCHED enthält, und die angegebenen Spark- und Iceberg-Versionen verwendet werden. Lies die Dokumentation zu Cost of Retries und verfolge, wie das Join-Ergebnis zur Laufzeit die Auswahl zwischen dem Append- und dem Overwrite-Verhalten bestimmt; fertig ist die Aufgabe, wenn gleichzeitige No-Match-Merges keine doppelten Datensätze erzeugen und das Verhalten durch Regressionstests abgedeckt ist.
Vom Indexierungsmodell aus dem Issue-Text verfasst.
Bewertung
- Tech-Stack
- java, spark, sql
- Bereich
- data-engineering, databases, distributed-systems
- Issue-Typ
- Bug
- Schwierigkeit
- 4/5
- Geschätzter Aufwand
- 3-5 Tage
- Aktivitätsstatus
- Ruhig
- Klarheit
- Größtenteils klar
- Anfängerfreundlichkeit
- 48/100