inject v2

Author: cxzl25

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

@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
 object PaimonDeleteTable extends Rule[LogicalPlan] with RowLevelHelper {
 
   /** Determines if DataSourceV2 delete is not supported for the given table. */
-  private def shouldFallbackToV1Delete(table: SparkTable, condition: Expression): Boolean = {
+  def shouldFallbackToV1Delete(table: SparkTable, condition: Expression): Boolean = {
     val baseTable = table.getTable
     org.apache.spark.SPARK_VERSION < "3.5" ||
     !baseTable.isInstanceOf[FileStoreTable] ||

paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala

@@ -19,6 +19,7 @@
 package org.apache.spark.sql.catalyst.analysis
 
 import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.catalyst.analysis.PaimonDeleteTable
 
 import org.apache.spark.sql.catalyst.expressions.{Alias, EqualNullSafe, Expression, Literal, Not}
 import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
@@ -49,7 +50,9 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand {
 
         case r @ ExtractV2Table(t: SupportsRowLevelOperations)
             if !t.isInstanceOf[SparkTable] || (t
-              .isInstanceOf[SparkTable] && t.asInstanceOf[SparkTable].useV2Write) =>
+              .isInstanceOf[SparkTable] && !PaimonDeleteTable.shouldFallbackToV1Delete(
+              t.asInstanceOf[SparkTable],
+              cond)) =>
           val table = buildOperationTable(t, DELETE, CaseInsensitiveStringMap.empty())
           table.operation match {
             case _: SupportsDelta =>