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 + ) + } + } }