[spark] Fix missing data for delete hiding remaining records

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

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.commands
 
+import org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW
 import org.apache.paimon.Snapshot
 import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
 import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
@@ -104,8 +105,13 @@ case class DeleteFromPaimonTableCommand(
         data = selectWithRowTracking(data)
       }
 
-      // only write new files, should have no compaction
-      val addCommitMessage = writer.writeOnly().withRowTracking().write(data)
+      val rewriteWriter =
+        if (coreOptions.mergeEngine() == FIRST_ROW) {
+          writer.withIgnorePreviousFiles()
+        } else {
+          writer.writeOnly()
+        }
+      val addCommitMessage = rewriteWriter.withRowTracking().write(data)
 
       // Step5: convert the deleted files that need to be written to commit message.
       val deletedCommitMessage = buildDeletedCommitMessage(touchedFiles)

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

@@ -57,7 +57,8 @@ import scala.collection.JavaConverters._
 case class PaimonSparkWriter(
     table: FileStoreTable,
     writeRowTracking: Boolean = false,
-    batchId: Option[Long] = None)
+    batchId: Option[Long] = None,
+    ignorePreviousFiles: Boolean = false)
   extends WriteHelper {
 
   private lazy val tableSchema = table.schema
@@ -115,9 +116,13 @@ case class PaimonSparkWriter(
     PaimonSparkWriter(table.copy(singletonMap(WRITE_ONLY.key(), "true")))
   }
 
+  def withIgnorePreviousFiles(): PaimonSparkWriter = {
+    copy(ignorePreviousFiles = true)
+  }
+
   def withRowTracking(): PaimonSparkWriter = {
     if (coreOptions.rowTrackingEnabled()) {
-      PaimonSparkWriter(table, writeRowTracking = true)
+      PaimonSparkWriter(table, writeRowTracking = true, ignorePreviousFiles = ignorePreviousFiles)
     } else {
       this
     }
@@ -175,7 +180,8 @@ case class PaimonSparkWriter(
         fullCompactionDeltaCommits,
         batchId,
         uriReaderFactory,
-        postponePartitionBucketComputer
+        postponePartitionBucketComputer,
+        ignorePreviousFiles
       )
 
     def sparkParallelism = {

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala

@@ -38,7 +38,8 @@ case class PaimonDataWrite(
     fullCompactionDeltaCommits: Option[Int],
     batchId: Option[Long],
     uriReaderFactory: UriReaderFactory,
-    postponePartitionBucketComputer: Option[BinaryRow => Integer])
+    postponePartitionBucketComputer: Option[BinaryRow => Integer],
+    ignorePreviousFiles: Boolean = false)
   extends abstractInnerTableDataWrite[Row]
   with InnerTableV1DataWrite {
 
@@ -47,6 +48,9 @@ case class PaimonDataWrite(
   val write: TableWriteImpl[Row] = {
     val _write = writeBuilder.newWrite().asInstanceOf[TableWriteImpl[Row]]
     _write.withIOManager(ioManager)
+    if (ignorePreviousFiles) {
+      _write.withIgnorePreviousFiles(true)
+    }
     if (writeRowTracking) {
       _write.withWriteType(writeType)
     }

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

@@ -252,6 +252,21 @@ abstract class DeleteFromTableTestBase extends PaimonSparkTestBase {
       }
   }
 
+  test("Paimon Delete: first-row table") {
+    withTable("t") {
+      sql("""CREATE TABLE t (id INT, name STRING)
+            |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1', 'merge-engine' = 'first-row')
+            |""".stripMargin)
+      sql("INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+      checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(1, "a"), Row(2, "b"), Row(3, "c")))
+
+      sql("DELETE FROM t WHERE id = 3")
+
+      checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(1, "a"), Row(2, "b")))
+    }
+  }
+
   test(s"test delete with primary key") {
     spark.sql(
       s"""