[spark] Fix MERGE INTO silently skipping WHEN NOT MATCHED BY SOURCE actions

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

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

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