[spark] Version-conditional Resolution rule for Spark 4.1 row-level rewrites

Spark 4.1 moved its built-in RewriteUpdateTable / RewriteDeleteFromTable /
RewriteMergeIntoTable into the Resolution batch, and the Resolution batch
marks the plan analyzed=true before postHoc rules run. Diagnostic runs on
PR 7648 confirmed that for Spark 4.1 + useV2Write=false + append-only tables,
PaimonUpdateTable's postHoc rule matches, takes the V1 fallback branch, and
returns UpdatePaimonTableCommand, yet the physical planner still sees the
original UpdateTable node and fails with UNSUPPORTED_FEATURE.TABLE_OPERATION.
The transformed plan is silently dropped in Spark 4.1's analyzed-plan flow.

Register PaimonUpdateTable / PaimonDeleteTable / PaimonMergeInto as Resolution
rules ONLY on Spark 4.1+, so their transform output sticks. Spark 3.x keeps
the original postHoc-only registration — it has no competing Resolution-batch
rewrite rule and the postHoc path works as before.

Inside the rules, skip tables that expose SupportsRowLevelOperations when on
Spark 4.1+. Those tables already go through Spark's built-in V2 row-level
rewrite; hijacking them here would force V1 fallback and drop V2-only
signals such as `_ROW_ID`, tripping CheckAnalysis with MISSING_ATTRIBUTES.
The guard is version-gated so Spark 3.2/3.3 (no built-in V2 merge rewrite)
keeps using Paimon's rule for useV2Write=true tables, avoiding the regression
that appeared when the guard was unconditional (Round 6 on PR 7648).

Author: kerwin-zk

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

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.catalyst.analysis
 
+import org.apache.paimon.spark.SparkTableWithRowLevelOps
 import org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand
 import org.apache.paimon.table.FileStoreTable
 
@@ -36,10 +37,13 @@ object PaimonDeleteTable extends Rule[LogicalPlan] with RowLevelHelper {
     // (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 d @ DeleteFromTable(PaimonRelation(table), condition)
-            if d.resolved && shouldFallbackToV1Delete(table, condition) =>
+            if d.resolved && (!skipV2RowLevelTables ||
+              !table.isInstanceOf[SparkTableWithRowLevelOps]) &&
+              shouldFallbackToV1Delete(table, condition) =>
           checkPaimonTable(table.getTable)
 
           table.getTable match {

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/catalyst/analysis/PaimonUpdateTable.scala

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.catalyst.analysis
 
+import org.apache.paimon.spark.SparkTableWithRowLevelOps
 import org.apache.paimon.spark.commands.UpdatePaimonTableCommand
 import org.apache.paimon.table.FileStoreTable
 
@@ -42,9 +43,18 @@ object PaimonUpdateTable
     // `transformDown` (which unconditionally visits every node) guarded by
     // `AnalysisHelper.allowInvokingTransformsInAnalyzer` so the in-analyzer assertion does not
     // trip. The pattern guard keeps the rewrite restricted to fully resolved plans.
+    // On Spark 4.1 this rule is also registered as a resolutionRule (postHoc output is silently
+    // discarded by Spark 4.1's analyzed-plan handling). When that happens, skip tables exposing
+    // SupportsRowLevelOperations because Spark's built-in RewriteUpdateTable will handle the V2
+    // row-level path for them and any hijack here would drop V2-only signals such as `_ROW_ID`
+    // and cause CheckAnalysis to reject the plan with MISSING_ATTRIBUTES. Older Spark versions
+    // have no competing RewriteUpdateTable, so the rule still fires for every Paimon relation.
+    val skipV2RowLevelTables = org.apache.spark.SPARK_VERSION >= "4.1"
     AnalysisHelper.allowInvokingTransformsInAnalyzer {
       plan.transformDown {
-        case u @ UpdateTable(PaimonRelation(table), assignments, condition) if u.resolved =>
+        case u @ UpdateTable(PaimonRelation(table), assignments, condition)
+            if u.resolved && (!skipV2RowLevelTables ||
+              !table.isInstanceOf[SparkTableWithRowLevelOps]) =>
           checkPaimonTable(table.getTable)
 
           table.getTable match {

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

@@ -51,6 +51,21 @@ class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     extensions.injectPostHocResolutionRule(_ => PaimonDeleteTable)
     extensions.injectPostHocResolutionRule(spark => PaimonMergeInto(spark))
 
+    // 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
     PaimonTableValuedFunctions.supportedFnNames.foreach {
       fnName =>