[spark] Add point-delete fast path for primary key DELETE to skip the target table scan

docs/generated/spark_connector_configuration.html

@@ -26,6 +26,18 @@
         </tr>
     </thead>
     <tbody>
+        <tr>
+            <td><h5>delete.point-delete.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enable the point-delete fast path for primary key DELETE. When enabled, a DELETE whose condition pins all primary key columns (literals or a subquery) writes -D rows directly without scanning the target table. Note: such -D rows carry NULL for non-key columns, so incremental/audit_log reads will not see the old field values of deleted rows.</td>
+        </tr>
+        <tr>
+            <td><h5>delete.point-delete.max-rows</h5></td>
+            <td style="word-wrap: break-word;">1000000</td>
+            <td>Long</td>
+            <td>Max number of -D rows the primary-key point-delete fast path may build on the driver; beyond this, DELETE falls back to the scan-based path.</td>
+        </tr>
         <tr>
             <td><h5>legacy-timestamp-mapping.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

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

@@ -47,6 +47,24 @@ public class SparkConnectorOptions {
                             "Parallelism used to repartition a single-partition LIMIT input before "
                                     + "executing a lateral vector search.");
 
+    public static final ConfigOption<Boolean> DELETE_POINT_DELETE_ENABLED =
+            key("delete.point-delete.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to enable the point-delete fast path for primary key DELETE. When enabled, "
+                                    + "a DELETE whose condition pins all primary key columns (literals or a subquery) "
+                                    + "writes -D rows directly without scanning the target table. Note: such -D rows "
+                                    + "carry NULL for non-key columns, so incremental/audit_log reads will not see the "
+                                    + "old field values of deleted rows.");
+
+    public static final ConfigOption<Long> DELETE_POINT_DELETE_MAX_ROWS =
+            key("delete.point-delete.max-rows")
+                    .longType()
+                    .defaultValue(1000000L)
+                    .withDescription(
+                            "Max number of -D rows the primary-key point-delete fast path may build on the driver; beyond this, DELETE falls back to the scan-based path.");
+
     public static final ConfigOption<Boolean> MERGE_SCHEMA =
             key("write.merge-schema")
                     .booleanType()

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

@@ -18,20 +18,29 @@
 
 package org.apache.paimon.spark.commands
 
+import org.apache.paimon.CoreOptions.MergeEngine
 import org.apache.paimon.Snapshot
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions
+import org.apache.paimon.spark.SparkConnectorOptions
 import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
 import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
+import org.apache.paimon.spark.util.OptionUtils
 import org.apache.paimon.table.FileStoreTable
 import org.apache.paimon.table.PrimaryKeyTableUtils.validatePKUpsertDeletable
 import org.apache.paimon.table.sink.CommitMessage
 import org.apache.paimon.types.RowKind
 
-import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.{DataFrame, Row, SparkSession}
 import org.apache.spark.sql.PaimonUtils.createDataset
-import org.apache.spark.sql.catalyst.expressions.{EqualNullSafe, Expression, Literal, Not}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualNullSafe, EqualTo, Expression, In, InSet, InSubquery, ListQuery, Literal, Not}
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, SupportsSubquery}
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.functions.{col, lit}
+import org.apache.spark.sql.types.{DataType, StructField, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
 
 case class DeleteFromPaimonTableCommand(
     relation: DataSourceV2Relation,
@@ -41,6 +50,10 @@ case class DeleteFromPaimonTableCommand(
   with ExpressionHelper
   with SupportsSubquery {
 
+  // Guards cartesian blow-up of multi-column IN lists on the driver.
+  private def pointDeleteMaxRows: Long =
+    OptionUtils.getOptionString(SparkConnectorOptions.DELETE_POINT_DELETE_MAX_ROWS).toLong
+
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val commitMessages = if (usePKUpsertDelete()) {
       performPrimaryKeyDelete(sparkSession)
@@ -60,12 +73,167 @@ case class DeleteFromPaimonTableCommand(
     }
   }
 
+  /**
+   * The fast path fills non-pk columns of the -D rows with NULL, which is only safe for DEDUPLICATE
+   * (the merge never reads fields of a delete row). Other engines (e.g. partial-update with
+   * remove-record-on-sequence-group) rely on field values of the delete row, and NOT NULL non-pk
+   * columns would reject the write. Fall back to scan otherwise. The fast path is opt-in because
+   * incremental/audit_log reads would see NULL instead of the old field values of deleted rows.
+   */
+  private def fastPathEligible: Boolean = {
+    OptionUtils.getOptionString(SparkConnectorOptions.DELETE_POINT_DELETE_ENABLED).toBoolean &&
+    coreOptions.mergeEngine() == MergeEngine.DEDUPLICATE && {
+      val pkSet = table.primaryKeys().asScala.toSet
+      table
+        .rowType()
+        .getFields
+        .asScala
+        .forall(f => pkSet.contains(f.name()) || f.`type`().isNullable)
+    } && {
+      // Sorted/global pk indexes are built from real field values; NULL-filled -D rows would
+      // leave the deletion invisible to index lookups, so fall back to the scan path.
+      PrimaryKeyIndexDefinitions.create(table.schema()).definitions().isEmpty
+    }
+  }
+
   private def performPrimaryKeyDelete(sparkSession: SparkSession): Seq[CommitMessage] = {
-    val df = createDataset(sparkSession, Filter(condition, relation))
+    // Fast path: when the matched keys are fully described by the condition itself — literals
+    // (pk = v / pk IN (...)) or a pk IN (subquery) — build -D rows without scanning the target
+    // table. Absent keys are harmless (merged away in compaction). Otherwise fall back to scan.
+    val keyDf = if (!fastPathEligible) {
+      None
+    } else {
+      extractPointDeleteKeys()
+        .map(literalKeyDataFrame(sparkSession, _))
+        .orElse(extractSubqueryKeyDataFrame(sparkSession))
+    }
+    keyDf match {
+      case Some(keys) =>
+        writer.write(buildDeleteDataFrame(keys))
+      case None =>
+        val df = createDataset(sparkSession, Filter(condition, relation))
+          .withColumn(ROW_KIND_COL, lit(RowKind.DELETE.toByteValue))
+        writer.write(df)
+    }
+  }
+
+  /** pk IN (subquery) covering all pk columns -> key DataFrame from the subquery only. */
+  private def extractSubqueryKeyDataFrame(sparkSession: SparkSession): Option[DataFrame] = {
+    val primaryKeys = table.primaryKeys().asScala.toSeq
+    condition match {
+      case InSubquery(values, ListQuery(plan, _, _, _, _, _))
+          if primaryKeys.nonEmpty && values.size == primaryKeys.size &&
+            values.forall(_.isInstanceOf[Attribute]) =>
+        val resolver = conf.resolver
+        val valueNames = values.map(_.asInstanceOf[Attribute].name)
+        val coversAllPks = primaryKeys.forall(pk => valueNames.exists(resolver(_, pk))) &&
+          valueNames.forall(n => primaryKeys.exists(resolver(n, _)))
+        if (coversAllPks) {
+          // Subquery output columns correspond positionally to `values`; rename to pk names.
+          val keyDf = createDataset(sparkSession, plan)
+          Some(keyDf.toDF(valueNames: _*))
+        } else {
+          None
+        }
+      case _ => None
+    }
+  }
+
+  /** Materializes extracted literal key rows into a small local DataFrame. */
+  private def literalKeyDataFrame(
+      sparkSession: SparkSession,
+      keyRows: Seq[Map[String, Any]]): DataFrame = {
+    val tableSchema = StructType(
+      relation.output.map(a => StructField(a.name, a.dataType, a.nullable)))
+    val pkSchema = StructType(tableSchema.fields.filter(f => keyRows.head.contains(f.name)))
+    val sparkRows =
+      keyRows.map(m => Row.fromSeq(pkSchema.fields.map(f => convertLiteral(m(f.name), f.dataType))))
+    sparkSession.createDataFrame(sparkSession.sparkContext.parallelize(sparkRows, 1), pkSchema)
+  }
+
+  /**
+   * Extracts key rows when the condition is a conjunction of pk = literal / pk IN (literals)
+   * covering all pk columns; None -> fall back to the scan path.
+   */
+  private def extractPointDeleteKeys(): Option[Seq[Map[String, Any]]] = {
+    val primaryKeys = table.primaryKeys().asScala.toSeq
+    if (condition == null || primaryKeys.isEmpty) {
+      None
+    } else {
+      val resolver = conf.resolver
+      // pk column -> distinct literal values; matched stays true only if every conjunct fits
+      val keyValues = mutable.LinkedHashMap.empty[String, Seq[Any]]
+      var matched = true
+
+      def pkName(attr: Attribute): Option[String] =
+        primaryKeys.find(pk => resolver(attr.name, pk))
+
+      def splitAnd(e: Expression): Seq[Expression] = e match {
+        case And(l, r) => splitAnd(l) ++ splitAnd(r)
+        case other => Seq(other)
+      }
+
+      def record(attr: Attribute, values: Seq[Any]): Unit = {
+        // NULL literals never match under SQL three-valued logic; let the scan path handle them.
+        if (values.exists(_ == null)) {
+          matched = false
+        } else {
+          pkName(attr) match {
+            case Some(pk) if !keyValues.contains(pk) => keyValues(pk) = values
+            case _ => matched = false
+          }
+        }
+      }
+
+      splitAnd(condition).foreach {
+        case _ if !matched => // short-circuit remaining conjuncts
+        case EqualTo(attr: Attribute, Literal(v, _)) => record(attr, Seq(v))
+        case EqualTo(Literal(v, _), attr: Attribute) => record(attr, Seq(v))
+        case In(attr: Attribute, values) if values.forall(_.isInstanceOf[Literal]) =>
+          record(attr, values.map(_.asInstanceOf[Literal].value).distinct)
+        case InSet(attr: Attribute, values) => record(attr, values.toSeq)
+        case _ => matched = false
+      }
+
+      lazy val totalRows = keyValues.values.map(_.size.toLong).product
+      if (!matched || primaryKeys.exists(pk => !keyValues.contains(pk)) || totalRows <= 0) {
+        None
+      } else if (totalRows > pointDeleteMaxRows) {
+        logInfo(
+          s"Point-delete rows $totalRows exceeds ${SparkConnectorOptions.DELETE_POINT_DELETE_MAX_ROWS.key()}" +
+            s"=$pointDeleteMaxRows, falling back to scan-based delete; consider IN (subquery).")
+        None
+      } else {
+        // cartesian product of per-column value lists
+        var rows: Seq[Map[String, Any]] = Seq(Map.empty)
+        for ((pk, values) <- keyValues) {
+          rows = for (row <- rows; v <- values) yield row + (pk -> v)
+        }
+        Some(rows)
+      }
+    }
+  }
+
+  /** Builds the -D DataFrame from a key DataFrame without reading the target table. */
+  private def buildDeleteDataFrame(keyDf: DataFrame): DataFrame = {
+    val keyCols = keyDf.schema.fieldNames.toSet
+    val projected = relation.output.map {
+      a =>
+        if (keyCols.contains(a.name)) {
+          col(a.name).cast(a.dataType).as(a.name)
+        } else {
+          lit(null).cast(a.dataType).as(a.name)
+        }
+    }
+    keyDf
+      .select(projected: _*)
       .withColumn(ROW_KIND_COL, lit(RowKind.DELETE.toByteValue))
-    writer.write(df)
   }
 
+  /** Catalyst literal internal values -> external row values accepted by createDataFrame. */
+  private def convertLiteral(v: Any, dataType: DataType): Any =
+    CatalystTypeConverters.convertToScala(v, dataType)
+
   private def performNonPrimaryKeyDelete(sparkSession: SparkSession): Seq[CommitMessage] = {
     val readSnapshot = table.snapshotManager().latestSnapshot()
     // Step1: the candidate data splits which are filtered by Paimon Predicate.