[spark] Eliminate the union stage when merging into without notMatchedActions

Author: Aitozi

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala

@@ -144,29 +144,38 @@ case class MergeIntoPaimonTable(
       }
     } else {
       val touchedFilePathsSet = mutable.Set.empty[String]
+      val intersectionFilePaths = mutable.Set.empty[String]
+
       def hasUpdate(actions: Seq[MergeAction]): Boolean = {
         actions.exists {
           case _: UpdateAction | _: DeleteAction => true
           case _ => false
         }
       }
-      if (hasUpdate(matchedActions) || notMatchedActions.nonEmpty) {
-        touchedFilePathsSet ++= findTouchedFiles(
-          targetDS.alias("_left").join(sourceDS, toColumn(mergeCondition), "inner"),
+
+      def findTouchedFiles0(joinType: String): Array[String] = {
+        findTouchedFiles(
+          targetDS.alias("_left").join(sourceDS, toColumn(mergeCondition), joinType),
           sparkSession,
-          "_left." + FILE_PATH_COLUMN
-        )
+          "_left." + FILE_PATH_COLUMN)
       }
+
+      if (hasUpdate(matchedActions)) {
+        touchedFilePathsSet ++= findTouchedFiles0("inner")
+      } else if (notMatchedActions.nonEmpty) {
+        intersectionFilePaths ++= findTouchedFiles0("inner")
+      }
+
       if (hasUpdate(notMatchedBySourceActions)) {
-        touchedFilePathsSet ++= findTouchedFiles(
-          targetDS.alias("_left").join(sourceDS, toColumn(mergeCondition), "left_anti"),
-          sparkSession,
-          "_left." + FILE_PATH_COLUMN)
+        touchedFilePathsSet ++= findTouchedFiles0("left_anti")
       }
 
-      val targetFilePaths: Array[String] = findTouchedFiles(targetDS, sparkSession)
       val touchedFilePaths: Array[String] = touchedFilePathsSet.toArray
-      val unTouchedFilePaths = targetFilePaths.filterNot(touchedFilePaths.contains)
+      val unTouchedFilePaths = if (notMatchedActions.nonEmpty) {
+        intersectionFilePaths.diff(touchedFilePathsSet).toArray
+      } else {
+        Array[String]()
+      }
 
       val (touchedFiles, touchedFileRelation) =
         createNewRelation(touchedFilePaths, dataFilePathToMeta, relation)
@@ -177,13 +186,9 @@ case class MergeIntoPaimonTable(
       // modified and was from touched file, it should be kept too.
       val touchedDsWithFileTouchedCol = createDataset(sparkSession, touchedFileRelation)
         .withColumn(FILE_TOUCHED_COL, lit(true))
-      val targetDSWithFileTouchedCol = if (notMatchedBySourceActions.nonEmpty) {
-        touchedDsWithFileTouchedCol.union(
-          createDataset(sparkSession, unTouchedFileRelation)
-            .withColumn(FILE_TOUCHED_COL, lit(false)))
-      } else {
-        touchedDsWithFileTouchedCol
-      }
+      val targetDSWithFileTouchedCol = touchedDsWithFileTouchedCol.union(
+        createDataset(sparkSession, unTouchedFileRelation)
+          .withColumn(FILE_TOUCHED_COL, lit(false)))
 
       val toWriteDS =
         constructChangedRows(sparkSession, targetDSWithFileTouchedCol).drop(ROW_KIND_COL)