You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Anupam Yadav (Jira)" <ji...@apache.org> on 2026/05/13 20:12:00 UTC
[jira] [Commented] (SPARK-56683) MERGE INTO TABLE reads the source twice and the two reads can disagree leading to data inconsistency
[ https://issues.apache.org/jira/browse/SPARK-56683?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18080761#comment-18080761 ]
Anupam Yadav commented on SPARK-56683:
--------------------------------------
I'd be interested in working on a fix for this. A few questions on approach:
The simplest path I see is adding source materialization for DSv2 MERGE, similar to what Delta does with `MergeIntoMaterializeSource`.
This could be scoped as:
1. Detect when the source plan is non-deterministic (or always materialize as a safe default)
2. Insert a materialization step before `RewriteMergeIntoTable` splits the source into two references
Alternatively, the more general solution in SPARK-56685 (guaranteed CTE reuse) would fix this at the Catalyst level for all cases, but that's a
larger change.
Would a targeted DSv2 materialization approach be acceptable, or is the preference to solve this at the Catalyst level? Happy to start with a
design doc or prototype PR if there's interest.
> MERGE INTO TABLE reads the source twice and the two reads can disagree leading to data inconsistency
> ----------------------------------------------------------------------------------------------------
>
> Key: SPARK-56683
> URL: https://issues.apache.org/jira/browse/SPARK-56683
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0
> Reporter: Juliusz Sompolski
> Priority: Major
>
> RewriteMergeIntoTable rewrites a MERGE INTO statement into a plan that references the source query in two positions: once as the streamed input to the join that pairs source rows with target rows, and once inside a subquery that the rewrite uses to identify which rows or groups have matching source rows.
> The two positions are independent reads of the same source. When the source is non-deterministic — for example, a table with concurrent writers, a streaming source, or a query containing expressions like rand() — the two reads can observe different sets of rows. The MERGE result is then computed against an inconsistent picture of the source: rows can be filtered in or out by the subquery while the join sees a different set of rows, producing dropped, duplicated, or wrongly-matched rows.
> The two reads of the source need to be made consistent so that both positions in the rewritten plan see the same source data, regardless of source determinism.
> Delta Lake resolved this with [https://github.com/delta-io/delta/blob/master/spark/src/main/scala/org/apache/spark/sql/delta/commands/merge/MergeIntoMaterializeSource.scala] in their custom MERGE implementation, but DSv2 SupportsRowLevelOperation datasources suffer a possible data inconsistency issue because of it.
> This could be resolved with https://issues.apache.org/jira/browse/SPARK-56685
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org