[spark] Support V2 delta-based row-level operations for deletion-vector append tables

paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/SparkTable.scala

@@ -35,4 +35,6 @@ object SparkTable {
   def of(table: Table): SparkTable = SparkTable(table)
 
   private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean = false
+
+  def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
 }

paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/SparkTable.scala

@@ -36,4 +36,6 @@ object SparkTable {
   def of(table: Table): SparkTable = SparkTable(table)
 
   private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean = false
+
+  def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
 }

paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/SparkTable.scala

@@ -35,4 +35,6 @@ object SparkTable {
   def of(table: Table): SparkTable = SparkTable(table)
 
   private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean = false
+
+  def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = false
 }

paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonDeltaBatchWrite.scala

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.write
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.table.FileStoreTable
+
+import org.apache.spark.sql.connector.write.{DeltaBatchWrite, DeltaWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Spark 4.0 shadow of the `paimon-spark4-common` wrapper: identical source compiled against Spark
+ * 4.0.2 so the mixed-in `BatchWrite` methods carry no Spark 4.1-only `WriteSummary` signature. See
+ * [[PaimonDeltaWriteBase]].
+ */
+class PaimonDeltaBatchWrite(
+    table: FileStoreTable,
+    rowSchema: StructType,
+    rowIdSchema: StructType,
+    operationType: Snapshot.Operation,
+    readSnapshotId: Option[Long])
+  extends PaimonDeltaWriteBase(table, rowSchema, rowIdSchema, operationType, readSnapshotId)
+  with DeltaBatchWrite
+  with Serializable {
+
+  override def createBatchWriterFactory(info: PhysicalWriteInfo): DeltaWriterFactory =
+    createPaimonDeltaWriterFactory(info)
+
+  override def useCommitCoordinator(): Boolean = false
+
+  override def commit(messages: Array[WriterCommitMessage]): Unit = commitMessages(messages)
+
+  override def abort(messages: Array[WriterCommitMessage]): Unit = abortMessages(messages)
+}

paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala

@@ -25,7 +25,7 @@ import org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark4SqlExtensi
 import org.apache.paimon.spark.data.{Spark4ArrayData, Spark4InternalRow, Spark4InternalRowWithBlob, SparkArrayData, SparkInternalRow}
 import org.apache.paimon.spark.format.FormatTableBatchWrite
 import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
-import org.apache.paimon.spark.write.PaimonBatchWrite
+import org.apache.paimon.spark.write.{PaimonBatchWrite, PaimonDeltaBatchWrite}
 import org.apache.paimon.table.{FileStoreTable, FormatTable}
 import org.apache.paimon.types.{DataType, RowType}
 
@@ -220,6 +220,14 @@ class Spark4Shim extends SparkShim {
       copyOnWriteScan,
       operationType)
 
+  override def createPaimonDeltaBatchWrite(
+      table: FileStoreTable,
+      rowSchema: StructType,
+      rowIdSchema: StructType,
+      operationType: Snapshot.Operation,
+      readSnapshotId: Option[Long]): BatchWrite =
+    new PaimonDeltaBatchWrite(table, rowSchema, rowIdSchema, operationType, readSnapshotId)
+
   override def createFormatTableBatchWrite(
       table: FormatTable,
       overwriteDynamic: Option[Boolean],

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

@@ -40,6 +40,7 @@ object PaimonMetrics {
   val ADDED_TABLE_FILES = "addedTableFiles"
   val DELETED_TABLE_FILES = "deletedTableFiles"
   val INSERTED_RECORDS = "insertedRecords"
+  val UPDATED_RECORDS = "updatedRecords"
   val DELETED_RECORDS = "deletedRecords"
   val APPENDED_CHANGELOG_FILES = "appendedChangelogFiles"
   val PARTITIONS_WRITTEN = "partitionsWritten"
@@ -212,6 +213,15 @@ case class PaimonInsertedRecordsTaskMetric(value: Long) extends PaimonTaskMetric
   override def name(): String = PaimonMetrics.INSERTED_RECORDS
 }
 
+case class PaimonUpdatedRecordsMetric() extends PaimonSumMetric {
+  override def name(): String = PaimonMetrics.UPDATED_RECORDS
+  override def description(): String = "number of updated records"
+}
+
+case class PaimonUpdatedRecordsTaskMetric(value: Long) extends PaimonTaskMetric {
+  override def name(): String = PaimonMetrics.UPDATED_RECORDS
+}
+
 case class PaimonDeletedRecordsMetric() extends PaimonSumMetric {
   override def name(): String = PaimonMetrics.DELETED_RECORDS
   override def description(): String = "number of deleted records"

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

@@ -133,10 +133,25 @@ abstract class PaimonSparkTableBase(val table: Table)
       _metadataColumns.append(PaimonMetadataColumn.SEARCH_SCORE)
     }
 
+    // For tables on the delta-based row-level path these two columns form the delta row ID, and
+    // Spark rejects nullable row ID columns. Keep them nullable everywhere else: V1 commands
+    // project them through outer joins where the target side can be null (e.g. the not-matched
+    // rows of MERGE INTO).
+    val (filePathColumn, rowIndexColumn) =
+      if (
+        this.isInstanceOf[SparkTable] &&
+        SparkTable.supportsV2DeltaOps(this.asInstanceOf[SparkTable])
+      ) {
+        (
+          PaimonMetadataColumn.FILE_PATH.copy(nullable = false),
+          PaimonMetadataColumn.ROW_INDEX.copy(nullable = false))
+      } else {
+        (PaimonMetadataColumn.FILE_PATH, PaimonMetadataColumn.ROW_INDEX)
+      }
     _metadataColumns.appendAll(
       Seq(
-        PaimonMetadataColumn.FILE_PATH,
-        PaimonMetadataColumn.ROW_INDEX,
+        filePathColumn,
+        rowIndexColumn,
         PaimonMetadataColumn.PARTITION(partitionType),
         PaimonMetadataColumn.BUCKET
       ))

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

@@ -19,9 +19,9 @@
 package org.apache.paimon.spark
 
 import org.apache.paimon.spark.read.ObjectTableScanBuilder
-import org.apache.paimon.spark.rowops.PaimonSparkCopyOnWriteOperation
+import org.apache.paimon.spark.rowops.{PaimonSparkCopyOnWriteOperation, PaimonSparkDeltaOperation}
 import org.apache.paimon.table.`object`.ObjectTable
-import org.apache.paimon.table.{FileStoreTable, Table}
+import org.apache.paimon.table.{BucketMode, FileStoreTable, Table}
 
 import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsRowLevelOperations, TableCapability}
 import org.apache.spark.sql.connector.read.ScanBuilder
@@ -40,15 +40,17 @@ import java.util.{EnumSet => JEnumSet, Set => JSet}
  * If this base class implemented `SupportsRowLevelOperations`, Spark 4.1 would immediately call
  * `newRowLevelOperationBuilder` on tables whose V2 write is disabled (e.g. dynamic bucket or
  * primary-key tables that fall back to V1 write) and fail before Paimon has a chance to rewrite the
- * plan to a V1 command. Likewise, deletion-vector, data-evolution, and fixed-length CHAR tables
- * need to stay on Paimon's V1 postHoc path even when `useV2Write=true`, so they must also not
- * expose `SupportsRowLevelOperations`.
+ * plan to a V1 command. Likewise, data-evolution and fixed-length CHAR tables need to stay on
+ * Paimon's V1 postHoc path even when `useV2Write=true`, so they must also not expose
+ * `SupportsRowLevelOperations`.
  *
  * Tables that DO support V2 row-level operations use the [[SparkTableWithRowLevelOps]] subclass
  * instead; the [[SparkTable.of]] factory picks the right variant via
  * [[SparkTable.supportsV2RowLevelOps]]. Append-only tables, including row-tracking-only tables,
  * expose `SupportsRowLevelOperations` so DELETE, UPDATE, and MERGE INTO can go through the V2
- * copy-on-write path when the table has no PK, deletion vectors, data evolution, or CHAR columns.
+ * copy-on-write path when the table has no PK, deletion vectors, data evolution, or CHAR columns;
+ * unaware-bucket deletion-vector append tables expose it so DELETE can go through the delta path
+ * (see [[SparkTable.supportsV2DeltaOps]]).
  */
 case class SparkTable(override val table: Table) extends PaimonSparkTableBase(table)
 
@@ -64,7 +66,8 @@ class SparkTableWithRowLevelOps(tableArg: Table)
       info: RowLevelOperationInfo): RowLevelOperationBuilder = {
     table match {
       case t: FileStoreTable =>
-        () => new PaimonSparkCopyOnWriteOperation(t, info)
+        if (SparkTable.supportsV2DeltaOps(this)) { () => new PaimonSparkDeltaOperation(t, info) }
+        else { () => new PaimonSparkCopyOnWriteOperation(t, info) }
       case _ =>
         throw new UnsupportedOperationException(
           "Row-level write operation is only supported for FileStoreTable. " +
@@ -109,6 +112,10 @@ object SparkTable {
    * `NoSuchMethodError` at the first DML statement.
    */
   private[spark] def supportsV2RowLevelOps(sparkTable: SparkTable): Boolean = {
+    supportsV2CopyOnWriteOps(sparkTable) || supportsV2DeltaOps(sparkTable)
+  }
+
+  private def supportsV2CopyOnWriteOps(sparkTable: SparkTable): Boolean = {
     if (org.apache.spark.SPARK_VERSION < "3.5") return false
     if (!sparkTable.useV2Write) return false
     sparkTable.getTable match {
@@ -123,6 +130,34 @@ object SparkTable {
       case _ => false
     }
   }
+
+  /**
+   * Whether the table takes the delta-based row-level path ([[PaimonSparkDeltaOperation]]): an
+   * unaware-bucket append-only table with deletion vectors enabled. Bucketed append tables are
+   * excluded because commit conflict detection does not cover concurrent deletion-vector index
+   * changes of the same bucket (`IndexManifestFileHandler` overwrites the whole bucket's index);
+   * row-tracking deletion-vector tables are excluded until the delta path defines the row lineage
+   * semantics for rewritten rows.
+   *
+   * The Spark 3.2/3.3/3.4 shim `SparkTable` companions must expose a method with this exact
+   * signature (hard-coded `false`), same as [[supportsV2RowLevelOps]]. Public because the Spark 4.1
+   * rewrite scope (`PureAppendOnlyScope.targetsV2DeltaTable`) delegates to it as the single source
+   * of truth for the delta gating conditions.
+   */
+  def supportsV2DeltaOps(sparkTable: SparkTable): Boolean = {
+    if (org.apache.spark.SPARK_VERSION < "3.5") return false
+    if (!sparkTable.useV2Write) return false
+    sparkTable.getTable match {
+      case fs: FileStoreTable =>
+        fs.primaryKeys().isEmpty &&
+        fs.bucketMode() == BucketMode.BUCKET_UNAWARE &&
+        sparkTable.coreOptions.deletionVectorsEnabled() &&
+        !sparkTable.coreOptions.rowTrackingEnabled() &&
+        !sparkTable.coreOptions.dataEvolutionEnabled() &&
+        !SparkTypeUtils.containsCharType(fs.rowType())
+      case _ => false
+    }
+  }
 }
 
 case class SparkIcebergTable(table: Table) extends BaseTable

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkDeltaOperation.scala

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.rowops
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.Snapshot
+import org.apache.paimon.spark.PaimonScanBuilder
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH_COLUMN, ROW_INDEX_COLUMN}
+import org.apache.paimon.spark.write.PaimonDeltaWriteBuilder
+import org.apache.paimon.table.{FileStoreTable, InnerTable}
+
+import org.apache.spark.sql.connector.expressions.{Expressions, NamedReference}
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.connector.write.{DeltaWriteBuilder, LogicalWriteInfo, RowLevelOperation, RowLevelOperationInfo, SupportsDelta}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import java.util.{HashMap => JHashMap}
+
+/**
+ * A delta-based [[RowLevelOperation]] for deletion-vector enabled append tables: instead of
+ * rewriting whole files like [[PaimonSparkCopyOnWriteOperation]], deleted rows are marked in
+ * deletion vectors keyed by the `(__paimon_file_path, __paimon_row_index)` row ID.
+ *
+ * The scan side is a regular pushdown scan: Spark's runtime group filtering only applies to
+ * group-based (`ReplaceData`) plans, delta plans converge on matching rows through the rewrite
+ * condition and ordinary predicate pushdown.
+ */
+class PaimonSparkDeltaOperation(table: FileStoreTable, info: RowLevelOperationInfo)
+  extends RowLevelOperation
+  with SupportsDelta {
+
+  // Captured before the scan is planned. The deletion-vector maintainer must merge with the
+  // deletion vectors of the snapshot the job read, never a newer one: initializing it from the
+  // commit-time latest snapshot would silently absorb concurrent deletion-vector changes that
+  // commit conflict detection is supposed to reject (and an UPDATE/MERGE could resurrect a
+  // concurrently deleted row as its new version). Same contract as the V1 commands' readSnapshot.
+  private val readSnapshotId: Option[Long] =
+    Option(table.snapshotManager().latestSnapshot()).map(_.id())
+
+  override def command(): RowLevelOperation.Command = info.command()
+
+  override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
+    // Pin the scan to the captured snapshot so the read, the deletion-vector maintainer and the
+    // commit-time file mapping all observe the same version; a commit that lands in between (e.g.
+    // a concurrent compaction) then surfaces as a commit conflict instead of a spurious mismatch
+    // between the scanned files and the read snapshot.
+    val conf = new JHashMap[String, String](options.asCaseSensitiveMap())
+    readSnapshotId.foreach(id => conf.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), id.toString))
+    new PaimonScanBuilder(table.copy(conf).asInstanceOf[InnerTable])
+  }
+
+  override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder =
+    new PaimonDeltaWriteBuilder(
+      table,
+      info,
+      Snapshot.Operation.valueOf(command().toString),
+      readSnapshotId)
+
+  override def rowId(): Array[NamedReference] =
+    Array(Expressions.column(FILE_PATH_COLUMN), Expressions.column(ROW_INDEX_COLUMN))
+
+  // Paimon cannot update rows in place: an update is a deletion-vector mark on the old row plus
+  // an appended new row. Still answer false so that Spark plans a single UPDATE operation per row
+  // (no Expand doubling); the delta writer decomposes it internally.
+  override def representUpdateAsDeleteAndInsert(): Boolean = false
+
+  override def requiredMetadataAttributes(): Array[NamedReference] = Array.empty
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala

@@ -32,9 +32,16 @@ case class PaimonMetadataColumn(
     override val dataType: DataType,
     preserveOnDelete: Boolean = true,
     preserveOnUpdate: Boolean = true,
-    preserveOnReinsert: Boolean = false)
+    preserveOnReinsert: Boolean = false,
+    nullable: Boolean = true)
   extends PaimonMetadataColumnBase {
 
+  // Only affects the Spark `MetadataColumn` capability (thus the relation's metadata output and
+  // the delta row ID nullability check). `toStructField` / `toAttribute` stay nullable: V1
+  // commands project these columns through outer joins where the target side can be null, e.g.
+  // the not-matched rows of MERGE INTO.
+  override def isNullable: Boolean = nullable
+
   def toPaimonDataField: DataField = {
     new DataField(id, name, SparkTypeUtils.toPaimonType(dataType));
   }

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

@@ -21,19 +21,15 @@ package org.apache.paimon.spark.write
 import org.apache.paimon.Snapshot
 import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
 import org.apache.paimon.spark.{PaimonDeletedRecordsTaskMetric, SparkTypeUtils}
-import org.apache.paimon.spark.catalyst.Compatibility
 import org.apache.paimon.spark.commands.SparkDataFileMeta
 import org.apache.paimon.spark.metric.SparkMetricRegistry
 import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
 import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH, ROW_ID, SEQUENCE_NUMBER}
 import org.apache.paimon.table.{FileStoreTable, SpecialFields}
 import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage, CommitMessageImpl}
 
-import org.apache.spark.sql.PaimonSparkSession
 import org.apache.spark.sql.connector.metric.CustomTaskMetric
 import org.apache.spark.sql.connector.write.{DataWriterFactory, PhysicalWriteInfo, WriterCommitMessage}
-import org.apache.spark.sql.execution.SQLExecution
-import org.apache.spark.sql.execution.metric.SQLMetrics
 import org.apache.spark.sql.types.StructType
 
 import java.util.Collections
@@ -133,7 +129,9 @@ abstract class PaimonBatchWriteBase(
     } finally {
       batchTableCommit.close()
     }
-    postDriverMetrics(deletedRecordsTaskMetric(operation, addCommitMessage, deletedCommitMessage))
+    postDriverMetrics(
+      metricRegistry.buildSparkCommitMetrics() ++
+        deletedRecordsTaskMetric(operation, addCommitMessage, deletedCommitMessage))
     postCommit(commitMessages)
   }
 
@@ -178,24 +176,6 @@ abstract class PaimonBatchWriteBase(
     }
   }
 
-  // Spark support v2 write driver metrics since 4.0, see https://github.com/apache/spark/pull/48573
-  // To ensure compatibility with 3.x, manually post driver metrics here instead of using Spark's API.
-  protected def postDriverMetrics(extraMetrics: Array[CustomTaskMetric] = Array.empty): Unit = {
-    val spark = PaimonSparkSession.active
-    // todo: find a more suitable way to get metrics.
-    val commitMetrics = metricRegistry.buildSparkCommitMetrics() ++ extraMetrics
-    val executionId = spark.sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
-    val executionMetrics = Compatibility.getExecutionMetrics(spark, executionId.toLong).distinct
-    val metricUpdates = executionMetrics.flatMap {
-      m =>
-        commitMetrics.find(x => m.metricType.toLowerCase.contains(x.name.toLowerCase)) match {
-          case Some(customTaskMetric) => Some((m.accumulatorId, customTaskMetric.value()))
-          case None => None
-        }
-    }
-    SQLMetrics.postDriverMetricsUpdatedByValue(spark.sparkContext, executionId, metricUpdates)
-  }
-
   private def buildDeletedCommitMessage(
       deletedFiles: Seq[SparkDataFileMeta]): Seq[CommitMessage] = {
     logInfo(s"[V2 Write] Building deleted commit message for ${deletedFiles.size} files")