Revert "[spark] Revert MergeInto changes from Round 8; keep Update/Delete Resolution rule fix"

This reverts commit 432a263.

Author: kerwin-zk

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeIntoBase.scala

@@ -18,7 +18,7 @@
 
 package org.apache.paimon.spark.catalyst.analysis
 
-import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.{SparkTable, SparkTableWithRowLevelOps}
 import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
 import org.apache.paimon.spark.commands.{MergeIntoPaimonDataEvolutionTable, MergeIntoPaimonTable}
 
@@ -47,10 +47,16 @@ trait PaimonMergeIntoBase
     // `transformDown` (which unconditionally visits every node) guarded by
     // `AnalysisHelper.allowInvokingTransformsInAnalyzer` so the in-analyzer assertion does not
     // trip.
+    val skipV2RowLevelTables = org.apache.spark.SPARK_VERSION >= "4.1"
     AnalysisHelper.allowInvokingTransformsInAnalyzer {
       plan.transformDown {
         case merge: MergeIntoTable
-            if merge.resolved && PaimonRelation.isPaimonTable(merge.targetTable) =>
+            if merge.resolved && PaimonRelation.isPaimonTable(merge.targetTable) &&
+              (!skipV2RowLevelTables ||
+                !PaimonRelation
+                  .getPaimonRelation(merge.targetTable)
+                  .table
+                  .isInstanceOf[SparkTableWithRowLevelOps]) =>
           val relation = PaimonRelation.getPaimonRelation(merge.targetTable)
           val v2Table = relation.table.asInstanceOf[SparkTable]
           val dataEvolutionEnabled = v2Table.coreOptions.dataEvolutionEnabled()

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala

@@ -51,20 +51,19 @@ class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     extensions.injectPostHocResolutionRule(_ => PaimonDeleteTable)
     extensions.injectPostHocResolutionRule(spark => PaimonMergeInto(spark))
 
-    // Spark 4.1 additionally registers Update/Delete rules in the Resolution batch. Spark 4.1 moved
-    // its own RewriteUpdateTable / RewriteDeleteFromTable into the Resolution batch and marks the
-    // plan as analyzed=true before postHoc rules run. The postHoc rule's transformed output is then
-    // silently dropped for UpdateTable / DeleteFromTable nodes, leaving them in the plan for the
-    // physical planner to reject with UNSUPPORTED_FEATURE.TABLE_OPERATION. Running these rules in
-    // the Resolution batch lets Paimon's rewrite actually stick. Each rule skips tables advertising
-    // SupportsRowLevelOperations (gated on the same Spark 4.1 version check) so Spark's built-in
-    // rewrite owns the V2 row-level path for useV2Write=true tables. `PaimonMergeInto` is
-    // intentionally left out here — the postHoc path already works for merge plans (verified by
-    // Blob merge tests passing pre-Round-8 on PR 7648) and re-registering it in the Resolution
-    // batch introduced regressions on Blob tests.
+    // Spark 4.1 additionally registers the same rules in the Resolution batch. Spark 4.1 moved its
+    // own RewriteUpdateTable / RewriteDeleteFromTable / RewriteMergeIntoTable into the Resolution
+    // batch and marks the plan as analyzed=true before postHoc rules run. The postHoc rule's
+    // transformed output is then silently dropped, leaving the original UpdateTable / DeleteFromTable
+    // / MergeIntoTable in the plan for the physical planner to reject with
+    // UNSUPPORTED_FEATURE.TABLE_OPERATION. Running them in the Resolution batch lets Paimon's
+    // rewrite actually stick. The rules themselves skip tables advertising SupportsRowLevelOperations
+    // (gated on the same Spark 4.1 version check), so Spark's built-in rewrite still owns the V2
+    // row-level path for useV2Write=true tables.
     if (org.apache.spark.SPARK_VERSION >= "4.1") {
       extensions.injectResolutionRule(_ => PaimonUpdateTable)
       extensions.injectResolutionRule(_ => PaimonDeleteTable)
+      extensions.injectResolutionRule(spark => PaimonMergeInto(spark))
     }
 
     // table function extensions