diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java index 6e0bbf9c5897..c83ad8af5791 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java +++ b/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 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 MERGE_SCHEMA = key("write.merge-schema") .booleanType() 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..e09b7d018d2c 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 @@ -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. diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletePointFastPathTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletePointFastPathTest.scala new file mode 100644 index 000000000000..cfb0e576f70c --- /dev/null +++ b/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) + } +}