[spark] Refactor spark v2 DELETE (#6851)

Author: Zouxxyy

docs/layouts/shortcodes/generated/spark_connector_configuration.html

@@ -96,7 +96,7 @@
             <td><h5>write.use-v2-write</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>If true, v2 write will be used. Currently, only HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1 write for other bucket modes. Currently, Spark V2 write does not support TableCapability.STREAMING_WRITE and TableCapability.ACCEPT_ANY_SCHEMA.</td>
+            <td>If true, v2 write will be used. Currently, only HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1 write for other bucket modes. Currently, Spark V2 write does not support TableCapability.STREAMING_WRITE.</td>
         </tr>
     </tbody>
 </table>

paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala

@@ -18,4 +18,12 @@
 
 package org.apache.paimon.spark.sql
 
+import org.apache.spark.SparkConf
+
 class DeleteFromTableTest extends DeleteFromTableTestBase {}
+
+class V2DeleteFromTableTest extends DeleteFromTableTestBase {
+  override protected def sparkConf: SparkConf = {
+    super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+  }
+}

paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/V2DeleteFromTableTest.scala

@@ -18,4 +18,12 @@
 
 package org.apache.paimon.spark.sql
 
+import org.apache.spark.SparkConf
+
 class DeleteFromTableTest extends DeleteFromTableTestBase {}
+
+class V2DeleteFromTableTest extends DeleteFromTableTestBase {
+  override protected def sparkConf: SparkConf = {
+    super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+  }
+}

paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala

@@ -51,7 +51,7 @@ public class SparkConnectorOptions {
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
-                            "If true, v2 write will be used. Currently, only HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1 write for other bucket modes. Currently, Spark V2 write does not support TableCapability.STREAMING_WRITE and TableCapability.ACCEPT_ANY_SCHEMA.");
+                            "If true, v2 write will be used. Currently, only HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1 write for other bucket modes. Currently, Spark V2 write does not support TableCapability.STREAMING_WRITE.");
 
     public static final ConfigOption<Integer> MAX_FILES_PER_TRIGGER =
             key("read.stream.maxFilesPerTrigger")

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java

@@ -18,7 +18,8 @@
 
 package org.apache.paimon.spark
 
-import org.apache.paimon.table.Table
+import org.apache.paimon.spark.rowops.PaimonSparkCopyOnWriteOperation
+import org.apache.paimon.table.{FileStoreTable, Table}
 
 import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations
 import org.apache.spark.sql.connector.write.{RowLevelOperationBuilder, RowLevelOperationInfo}
@@ -27,9 +28,17 @@ import org.apache.spark.sql.connector.write.{RowLevelOperationBuilder, RowLevelO
 case class SparkTable(override val table: Table)
   extends PaimonSparkTableBase(table)
   with SupportsRowLevelOperations {
+
   override def newRowLevelOperationBuilder(
-      rowLevelOperationInfo: RowLevelOperationInfo): RowLevelOperationBuilder = {
-    new PaimonSparkRowLevelOperationBuilder(table, rowLevelOperationInfo)
+      info: RowLevelOperationInfo): RowLevelOperationBuilder = {
+    table match {
+      case t: FileStoreTable if useV2Write =>
+        () => new PaimonSparkCopyOnWriteOperation(t, info)
+      case _ =>
+        throw new UnsupportedOperationException(
+          s"Write operation is only supported for FileStoreTable with V2 write enabled. " +
+            s"Actual table type: ${table.getClass.getSimpleName}, useV2Write: $useV2Write")
+    }
   }
 }
 

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkCopyOnWriteOperation.scala

@@ -22,49 +22,33 @@ import org.apache.paimon.spark.SparkTable
 import org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand
 import org.apache.paimon.table.FileStoreTable
 
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
 import org.apache.spark.sql.catalyst.plans.logical.{DeleteFromTable, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.Rule
 
-import scala.collection.JavaConverters._
-
 object PaimonDeleteTable extends Rule[LogicalPlan] with RowLevelHelper {
 
-  /**
-   * Determines if DataSourceV2 delete is not supported for the given table. DataSourceV2 delete is
-   * not supported in the following scenarios:
-   *   - Spark version is 3.4 or earlier
-   *   - Table does not use V2 write
-   *   - Row tracking is enabled
-   *   - Deletion vectors are enabled
-   *   - Table has primary keys defined
-   *   - Table is not a FileStoreTable
-   *   - Data evolution is enabled
-   */
-  private def shouldFallbackToV1Delete(table: SparkTable): Boolean = {
+  /** Determines if DataSourceV2 delete is not supported for the given table. */
+  private def shouldFallbackToV1Delete(table: SparkTable, condition: Expression): Boolean = {
     val baseTable = table.getTable
-
-    val baseConditions = org.apache.spark.SPARK_VERSION <= "3.4" ||
-      !table.useV2Write ||
-      table.coreOptions.rowTrackingEnabled() ||
-      table.coreOptions.deletionVectorsEnabled() ||
-      !baseTable.primaryKeys().isEmpty
-
-    baseConditions || {
-      baseTable match {
-        case paimonTable: FileStoreTable =>
-          paimonTable.coreOptions().dataEvolutionEnabled()
-        case _ =>
-          true
-      }
-    }
+    org.apache.spark.SPARK_VERSION < "3.5" ||
+    !baseTable.isInstanceOf[FileStoreTable] ||
+    !baseTable.primaryKeys().isEmpty ||
+    !table.useV2Write ||
+    table.coreOptions.deletionVectorsEnabled() ||
+    table.coreOptions.rowTrackingEnabled() ||
+    table.coreOptions.dataEvolutionEnabled() ||
+    // todo: Optimize v2 delete when conditions are all partition filters
+    condition == null || condition == TrueLiteral
   }
 
   override val operation: RowLevelOp = Delete
 
   override def apply(plan: LogicalPlan): LogicalPlan = {
     plan.resolveOperators {
       case d @ DeleteFromTable(PaimonRelation(table), condition)
-          if d.resolved && shouldFallbackToV1Delete(table) =>
+          if d.resolved && shouldFallbackToV1Delete(table, condition) =>
         checkPaimonTable(table.getTable)
 
         table.getTable match {

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkRowLevelOperationBuilder.scala

@@ -92,28 +92,26 @@ case class PaimonFormatTableWriterBuilder(table: FormatTable, writeSchema: Struc
 
 private case class FormatTableBatchWrite(
     table: FormatTable,
-    overwriteDynamic: Boolean,
+    overwriteDynamic: Option[Boolean],
     overwritePartitions: Option[Map[String, String]],
     writeSchema: StructType)
   extends BatchWrite
   with Logging {
 
-  assert(
-    !(overwriteDynamic && overwritePartitions.exists(_.nonEmpty)),
-    "Cannot overwrite dynamically and by filter both")
-
   private val batchWriteBuilder = {
     val builder = table.newBatchWriteBuilder()
-    if (overwriteDynamic) {
+    // todo: add test for static overwrite the whole table
+    if (overwriteDynamic.contains(true)) {
       builder.withOverwrite()
     } else {
       overwritePartitions.foreach(partitions => builder.withOverwrite(partitions.asJava))
     }
     builder
   }
 
-  override def createBatchWriterFactory(info: PhysicalWriteInfo): DataWriterFactory =
-    FormatTableWriterFactory(batchWriteBuilder, writeSchema)
+  override def createBatchWriterFactory(info: PhysicalWriteInfo): DataWriterFactory = {
+    (_: Int, _: Long) => new FormatTableDataWriter(batchWriteBuilder, writeSchema)
+  }
 
   override def useCommitCoordinator(): Boolean = false
 
@@ -140,16 +138,6 @@ private case class FormatTableBatchWrite(
   }
 }
 
-private case class FormatTableWriterFactory(
-    batchWriteBuilder: BatchWriteBuilder,
-    writeSchema: StructType)
-  extends DataWriterFactory {
-
-  override def createWriter(partitionId: Int, taskId: Long): DataWriter[InternalRow] = {
-    new FormatTableDataWriter(batchWriteBuilder, writeSchema)
-  }
-}
-
 private class FormatTableDataWriter(batchWriteBuilder: BatchWriteBuilder, writeSchema: StructType)
   extends V2DataWrite
   with Logging {