[spark] Fix batch delete for partial update sequence groups (#9546)

Author: ArnavBalyan

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

@@ -18,7 +18,7 @@
 
 package org.apache.paimon.spark.commands
 
-import org.apache.paimon.Snapshot
+import org.apache.paimon.{CoreOptions, Snapshot}
 import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
 import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
 import org.apache.paimon.table.FileStoreTable
@@ -54,7 +54,8 @@ case class DeleteFromPaimonTableCommand(
   private def usePKUpsertDelete(): Boolean = {
     try {
       validatePKUpsertDeletable(table)
-      true
+      coreOptions.mergeEngine() != CoreOptions.MergeEngine.PARTIAL_UPDATE ||
+      coreOptions.toConfiguration.get(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE)
     } catch {
       case _: UnsupportedOperationException => false
     }

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala

@@ -509,6 +509,23 @@ abstract class DeleteFromTableTestBase extends PaimonSparkTestBase {
       }
   }
 
+  test("Paimon Delete: partial update with remove record on sequence group") {
+    spark.sql(s"""
+                 |CREATE TABLE T (id INT, g INT, v BIGINT)
+                 |TBLPROPERTIES (
+                 |  'primary-key' = 'id',
+                 |  'bucket' = '2',
+                 |  'merge-engine' = 'partial-update',
+                 |  'fields.g.sequence-group' = 'v',
+                 |  'partial-update.remove-record-on-sequence-group' = 'g')
+                 |""".stripMargin)
+
+    spark.sql("INSERT INTO T VALUES (1, 1, 10), (2, 1, 20)")
+    spark.sql("DELETE FROM T WHERE id = 1")
+
+    checkAnswer(spark.sql("SELECT * FROM T"), Row(2, 1, 20L))
+  }
+
   test("Paimon delete: non pk table commit kind") {
     for (dvEnabled <- Seq(true, false)) {
       withTable("t") {