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

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

@@ -39,6 +39,13 @@ public class SparkConnectorOptions {
                     .withDescription(
                             "If true, map Paimon TIMESTAMP to Spark TIMESTAMP instead of TIMESTAMP_NTZ.");
 
+    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

@@ -19,19 +19,26 @@
 package org.apache.paimon.spark.commands
 
 import org.apache.paimon.Snapshot
+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 +48,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)
@@ -61,11 +72,134 @@ case class DeleteFromPaimonTableCommand(
   }
 
   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 = 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 = {
+        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.

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

@@ -0,0 +1,157 @@
+/*
+ * 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.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+
+/** Tests for the DELETE point-delete fast path (pk fully pinned by literals -> no table scan). */
+class DeletePointFastPathTest extends PaimonSparkTestBase {
+
+  test("Point delete fast path: pk IN literals") {
+    spark.sql("""
+                |CREATE TABLE T (id INT, name STRING, v BIGINT)
+                |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO T VALUES (1,'a',10),(2,'b',20),(3,'c',30),(4,'d',40),(5,'e',50)")
+
+    // includes a non-existing key 99: harmless
+    spark.sql("DELETE FROM T WHERE id IN (2, 4, 99)")
+
+    checkAnswer(spark.sql("SELECT id FROM T ORDER BY id"), Row(1) :: Row(3) :: Row(5) :: Nil)
+  }
+
+  test("Point delete fast path: pk equality") {
+    spark.sql("""
+                |CREATE TABLE TEQ (id INT, v BIGINT)
+                |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TEQ VALUES (1,10),(2,20),(3,30)")
+
+    spark.sql("DELETE FROM TEQ WHERE id = 2")
+
+    checkAnswer(spark.sql("SELECT id FROM TEQ ORDER BY id"), Row(1) :: Row(3) :: Nil)
+  }
+
+  test("Point delete fast path: composite pk (partition + id)") {
+    spark.sql("""
+                |CREATE TABLE PT (id INT, name STRING, dt STRING)
+                |PARTITIONED BY (dt)
+                |TBLPROPERTIES ('primary-key' = 'dt,id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("""
+                |INSERT INTO PT VALUES
+                | (1,'a','p1'),(2,'b','p1'),(3,'c','p1'),
+                | (1,'x','p2'),(2,'y','p2')
+                |""".stripMargin)
+
+    // dt pinned by equality, id by IN -> covers full pk (dt, id)
+    spark.sql("DELETE FROM PT WHERE dt = 'p1' AND id IN (1, 3)")
+
+    checkAnswer(
+      spark.sql("SELECT id, dt FROM PT ORDER BY dt, id"),
+      Row(2, "p1") :: Row(1, "p2") :: Row(2, "p2") :: Nil)
+  }
+
+  test("Fallback: condition with non-pk column still works (scan path)") {
+    spark.sql("""
+                |CREATE TABLE TF (id INT, v BIGINT)
+                |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TF VALUES (1,10),(2,20),(3,30)")
+
+    // v is not a pk column -> must fall back to scan-based delete and still be correct
+    spark.sql("DELETE FROM TF WHERE id IN (1, 2) AND v > 15")
+
+    checkAnswer(spark.sql("SELECT id FROM TF ORDER BY id"), Row(1) :: Row(3) :: Nil)
+  }
+
+  test("Fallback: pk not fully pinned still works (scan path)") {
+    spark.sql("""
+                |CREATE TABLE TP (id INT, name STRING, dt STRING)
+                |PARTITIONED BY (dt)
+                |TBLPROPERTIES ('primary-key' = 'dt,id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TP VALUES (1,'a','p1'),(1,'x','p2'),(2,'b','p1')")
+
+    // only id pinned, dt free -> not a point delete, scan path deletes across partitions
+    spark.sql("DELETE FROM TP WHERE id = 1")
+
+    checkAnswer(spark.sql("SELECT id, dt FROM TP ORDER BY dt, id"), Row(2, "p1") :: Nil)
+  }
+
+  test("Point delete then re-insert same key") {
+    spark.sql("""
+                |CREATE TABLE TR (id INT, v BIGINT)
+                |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TR VALUES (1,10),(2,20)")
+    spark.sql("DELETE FROM TR WHERE id = 1")
+    spark.sql("INSERT INTO TR VALUES (1, 111)")
+
+    checkAnswer(spark.sql("SELECT * FROM TR ORDER BY id"), Row(1, 111L) :: Row(2, 20L) :: Nil)
+  }
+
+  test("Subquery fast path: pk IN (SELECT ...) from key table") {
+    spark.sql("""
+                |CREATE TABLE TS (id INT, name STRING, v BIGINT)
+                |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TS VALUES (1,'a',10),(2,'b',20),(3,'c',30),(4,'d',40),(5,'e',50)")
+
+    // key table with existing keys (2, 4) and a non-existing key (99)
+    spark.sql("CREATE TABLE SKEYS (id INT)")
+    spark.sql("INSERT INTO SKEYS VALUES (2), (4), (99)")
+
+    spark.sql("DELETE FROM TS WHERE id IN (SELECT id FROM SKEYS)")
+
+    checkAnswer(spark.sql("SELECT id FROM TS ORDER BY id"), Row(1) :: Row(3) :: Row(5) :: Nil)
+
+    spark.sql("CALL paimon.sys.compact(table => 'test.TS', compact_strategy => 'full')")
+    checkAnswer(spark.sql("SELECT id FROM TS ORDER BY id"), Row(1) :: Row(3) :: Row(5) :: Nil)
+  }
+
+  test("Subquery fast path: composite pk IN (SELECT ...) with renamed columns") {
+    spark.sql("""
+                |CREATE TABLE TS2 (id INT, name STRING, dt STRING)
+                |PARTITIONED BY (dt)
+                |TBLPROPERTIES ('primary-key' = 'dt,id', 'bucket' = '2',
+                | 'deletion-vectors.enabled' = 'true')
+                |""".stripMargin)
+    spark.sql("INSERT INTO TS2 VALUES (1,'a','p1'),(2,'b','p1'),(1,'x','p2'),(2,'y','p2')")
+
+    // key table columns named differently; aligned via SELECT aliases in the subquery
+    spark.sql("CREATE TABLE SKEYS2 (kid INT, kdt STRING)")
+    spark.sql("INSERT INTO SKEYS2 VALUES (1, 'p1'), (2, 'p2')")
+
+    spark.sql("DELETE FROM TS2 WHERE (dt, id) IN (SELECT kdt, kid FROM SKEYS2)")
+
+    checkAnswer(
+      spark.sql("SELECT id, dt FROM TS2 ORDER BY dt, id"),
+      Row(2, "p1") :: Row(1, "p2") :: Nil)
+  }
+}