diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala index 247a998e4752..e85ef0272ebb 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala +++ b/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) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala index b201b88110d9..35d7098529e6 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala +++ b/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 = { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala index af20144f5233..fd35b6cfa7d9 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala +++ b/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) } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala index 2b89afcc6827..90bddb7aae80 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala +++ b/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"""