From 85a8073baf7258ea08cb20121d835d11a232f9af Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Mon, 10 Aug 2026 18:31:19 +0800 Subject: [PATCH] [spark] Fix MERGE INTO silently skipping WHEN NOT MATCHED BY SOURCE actions `MergeIntoPaimonTable` extracts the target-only conjuncts of the merge condition and uses them to prune the target table before the full outer join (`filteredTargetPlan` / `targetOnlyCondition`). That pruning is sound for `WHEN MATCHED` and `WHEN NOT MATCHED`: a target row that fails a target-only conjunct can never satisfy the whole merge condition, so it can never be matched, and dropping it cannot change the outcome of those actions. It is not sound for `WHEN NOT MATCHED BY SOURCE`. The pruned-away rows are exactly the population that clause is defined over, so their actions are silently skipped -- no error, no warning, just fewer rows changed. For example, with a partitioned table and MERGE INTO target t USING source s ON t.a = s.a AND t.pt = 'p1' WHEN MATCHED THEN UPDATE SET t.b = s.b WHEN NOT MATCHED BY SOURCE THEN UPDATE SET t.c = 'stale' every row outside `pt = 'p1'` should be updated to `stale` (no source row can match it), but none of them is. The pruning was introduced together with MERGE INTO itself in #2331, one month before `WHEN NOT MATCHED BY SOURCE` was added in #2517, and its safety argument was never revisited. Note that the V2 row-level paths (`ReplaceData` / `WriteDelta`) are rewritten by Spark and are not affected, so the same statement currently produces different results depending on whether the table qualifies for `SparkTable.supportsV2RowLevelOps`. Primary key tables never qualify, so they always take the affected V1 path. This disables the pruning when the merge has any `WHEN NOT MATCHED BY SOURCE` action. Setting `targetOnlyCondition` to `None` covers all three places it feeds: `filteredTargetPlan`, `findCandidateDataSplits` and `targetDSWithFilePathCol`. A follow-up can restore part of the pruning by handling the excluded rows as a separate not-matched-by-source-only stream, which avoids joining them against the source while still applying their actions. --- .../spark/commands/MergeIntoPaimonTable.scala | 16 +++++++- .../spark/commands/MergeIntoPaimonTable.scala | 16 +++++++- .../sql/MergeIntoNotMatchedBySourceTest.scala | 37 +++++++++++++++++++ 3 files changed, 67 insertions(+), 2 deletions(-) diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala index c24d4d6675b9..d51504059df9 100644 --- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala +++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala @@ -60,8 +60,22 @@ case class MergeIntoPaimonTable( lazy val relation: DataSourceV2Relation = PaimonRelation.getPaimonRelation(targetTable) + /** + * The target-only part of the merge condition, used to prune the target table before the join. + * + * Pruning is only sound when the merge has no `WHEN NOT MATCHED BY SOURCE` action. A target row + * that fails the target-only condition can never satisfy the whole merge condition, so it can + * never be matched — dropping it therefore cannot change the outcome of the `WHEN MATCHED` and + * `WHEN NOT MATCHED` actions. Those very rows are, however, exactly the population that + * `WHEN NOT MATCHED BY SOURCE` is defined over, so pruning them would silently skip the actions + * that should apply to them. Disable the pruning in that case. + */ private lazy val (targetOnlyCondition, filteredTargetPlan): (Option[Expression], LogicalPlan) = { - val filtersOnlyTarget = getExpressionOnlyRelated(mergeCondition, targetTable) + val filtersOnlyTarget = if (notMatchedBySourceActions.isEmpty) { + getExpressionOnlyRelated(mergeCondition, targetTable) + } else { + None + } ( filtersOnlyTarget, filtersOnlyTarget diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala index c24d4d6675b9..d51504059df9 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala @@ -60,8 +60,22 @@ case class MergeIntoPaimonTable( lazy val relation: DataSourceV2Relation = PaimonRelation.getPaimonRelation(targetTable) + /** + * The target-only part of the merge condition, used to prune the target table before the join. + * + * Pruning is only sound when the merge has no `WHEN NOT MATCHED BY SOURCE` action. A target row + * that fails the target-only condition can never satisfy the whole merge condition, so it can + * never be matched — dropping it therefore cannot change the outcome of the `WHEN MATCHED` and + * `WHEN NOT MATCHED` actions. Those very rows are, however, exactly the population that + * `WHEN NOT MATCHED BY SOURCE` is defined over, so pruning them would silently skip the actions + * that should apply to them. Disable the pruning in that case. + */ private lazy val (targetOnlyCondition, filteredTargetPlan): (Option[Expression], LogicalPlan) = { - val filtersOnlyTarget = getExpressionOnlyRelated(mergeCondition, targetTable) + val filtersOnlyTarget = if (notMatchedBySourceActions.isEmpty) { + getExpressionOnlyRelated(mergeCondition, targetTable) + } else { + None + } ( filtersOnlyTarget, filtersOnlyTarget diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoNotMatchedBySourceTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoNotMatchedBySourceTest.scala index d2c822789ccf..0ab6e5755adf 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoNotMatchedBySourceTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoNotMatchedBySourceTest.scala @@ -220,4 +220,41 @@ trait MergeIntoNotMatchedBySourceTest extends PaimonSparkTestBase with PaimonTab } } } + + test("Paimon MergeInto: not matched by source is not narrowed by target-only condition") { + withTable("source", "target") { + + Seq((1, 100)).toDF("a", "b").createOrReplaceTempView("source") + + createTable("target", "a INT, b INT, c STRING, pt STRING", Seq("a", "pt"), Seq("pt")) + spark.sql(""" + |INSERT INTO target VALUES + | (1, 10, 'c1', 'p1'), (2, 20, 'c2', 'p1'), + | (3, 30, 'c3', 'p2'), (4, 40, 'c4', 'p2') + |""".stripMargin) + + // `t.pt = 'p1'` only references the target, so it is a candidate for pruning the target + // before the join. Pruning it away would also drop the 'p2' rows from the population that + // WHEN NOT MATCHED BY SOURCE is defined over, silently skipping their update. + spark.sql(""" + |MERGE INTO target t + |USING source s + |ON t.a = s.a AND t.pt = 'p1' + |WHEN MATCHED THEN + | UPDATE SET t.b = s.b + |WHEN NOT MATCHED BY SOURCE THEN + | UPDATE SET t.c = 'stale' + |""".stripMargin) + + // a=1: matched (pt='p1') => b updated to 100 + // a=2: not matched by source (pt='p1') => c = 'stale' + // a=3, a=4: not matched by source (pt='p2', excluded by the target-only condition, so no + // source row can ever match them) => c = 'stale' + checkAnswer( + spark.sql("SELECT a, b, c, pt FROM target ORDER BY a"), + Row(1, 100, "c1", "p1") :: Row(2, 20, "stale", "p1") :: + Row(3, 30, "stale", "p2") :: Row(4, 40, "stale", "p2") :: Nil + ) + } + } }