diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html index cd95fd5fd41a..3cf628e81119 100644 --- a/docs/generated/spark_connector_configuration.html +++ b/docs/generated/spark_connector_configuration.html @@ -26,6 +26,18 @@ + +
delete.point-delete.enabled
+ false + Boolean + 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. Such -D rows carry NULL for non-key columns, so the fast path only applies to the 'deduplicate' merge engine and only when the table has no 'sequence.field', no NOT NULL non-key column, no primary key index and no cross-partition update; otherwise DELETE falls back to the scan-based path. Note: incremental/audit_log reads will not see the old field values of rows deleted through the fast path. + + +
delete.point-delete.max-rows
+ 1000000 + Long + 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. +
legacy-timestamp-mapping.enabled
false 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 2f315b8df0f5..1a57a025add8 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 @@ -47,6 +47,27 @@ public class SparkConnectorOptions { "Parallelism used to repartition a single-partition LIMIT input before " + "executing a lateral vector search."); + public static final ConfigOption 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. Such -D rows carry " + + "NULL for non-key columns, so the fast path only applies to the 'deduplicate' " + + "merge engine and only when the table has no 'sequence.field', no NOT NULL " + + "non-key column, no primary key index and no cross-partition update; otherwise " + + "DELETE falls back to the scan-based path. Note: incremental/audit_log reads " + + "will not see the old field values of rows deleted through the fast path."); + + 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..9336043934fd 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,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,182 @@ 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. 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. + * + * It also requires that nothing else depends on those NULLs: + * - NOT NULL non-pk columns: the write would be rejected. + * - `sequence.field`: a NULL sequence value sorts oldest, so the delete would lose the merge. + * - partition columns outside the pk (cross partition update): the -D row would carry a NULL + * partition. + * - sorted/global pk indexes: they are built from real field values, so a NULL-filled -D row + * would leave the deletion invisible to index lookups. + */ + private def fastPathEligible: Boolean = { + OptionUtils.getOptionString(SparkConnectorOptions.DELETE_POINT_DELETE_ENABLED).toBoolean && + coreOptions.mergeEngine() == MergeEngine.DEDUPLICATE && + coreOptions.sequenceField().isEmpty && + !table.schema().crossPartitionUpdate() && { + val pkSet = table.primaryKeys().asScala.toSet + table + .rowType() + .getFields + .asScala + .forall(f => pkSet.contains(f.name()) || f.`type`().isNullable) + } && { + 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. Keys that do not exist only add a -D record that compaction removes later (it can + // however materialize a partition that did not exist before). 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 { + // `listQuery.children` are the outer references: a correlated subquery can not be planned on + // its own, so leave it to the scan path. Decorrelation also widens the subquery output, hence + // the output size check. + case InSubquery(values, listQuery: ListQuery) + if listQuery.children.isEmpty && primaryKeys.nonEmpty && + values.size == primaryKeys.size && values.forall(_.isInstanceOf[Attribute]) && + listQuery.plan.output.size == values.size => + 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, listQuery.plan).toDF(valueNames: _*) + // NULL never matches under SQL three-valued logic; dropping these rows keeps the fast + // path in sync with the scan path and avoids writing -D rows with NULL primary keys. + Some(keyDf.filter(valueNames.map(n => col(n).isNotNull).reduce(_ && _))) + } 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. 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..f43add4c1eab --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletePointFastPathTest.scala @@ -0,0 +1,365 @@ +/* + * 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 { + + private def totalRecordCount(tableName: String): Long = + spark + .sql(s"SELECT COALESCE(SUM(record_count), 0) FROM `$tableName$$files`") + .head() + .getLong(0) + + /** + * The fast path writes a -D row even for a key that does not exist, while the scan path finds + * nothing to delete and writes no record at all. So deleting an absent key tells the two paths + * apart. The table must be 'write-only' so that no compaction removes the record again. + */ + private def assertFastPath(tableName: String, condition: String): Unit = { + val before = totalRecordCount(tableName) + spark.sql(s"DELETE FROM $tableName WHERE $condition") + assert( + totalRecordCount(tableName) == before + 1, + s"expected the fast path to write one -D row for the absent key ($condition)") + } + + private def assertScanPath(tableName: String, condition: String): Unit = { + val before = totalRecordCount(tableName) + spark.sql(s"DELETE FROM $tableName WHERE $condition") + assert( + totalRecordCount(tableName) == before, + s"expected the scan path to write nothing for the absent key ($condition)") + } + + test("Point delete fast path: pk IN literals") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + 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) + } + } + + test("Fallback: NULL literal in condition follows SQL three-valued logic") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TNULL (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2') + |""".stripMargin) + spark.sql("INSERT INTO TNULL VALUES (1,10),(2,20)") + + // id = NULL / IN (..., NULL): NULL never matches, rows must NOT be deleted blindly + spark.sql("DELETE FROM TNULL WHERE id = NULL") + checkAnswer(spark.sql("SELECT count(*) FROM TNULL"), Row(2) :: Nil) + + spark.sql("DELETE FROM TNULL WHERE id IN (1, NULL)") + checkAnswer(spark.sql("SELECT id FROM TNULL"), Row(2) :: Nil) + } + } + + test("Fallback: table with NOT NULL non-pk column still works (scan path)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TNN (id INT, name STRING NOT NULL, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2') + |""".stripMargin) + spark.sql("INSERT INTO TNN VALUES (1,'a',10),(2,'b',20),(3,'c',30)") + + // fast path would write NULL into the NOT NULL column; must fall back and still succeed + spark.sql("DELETE FROM TNN WHERE id IN (1, 3)") + checkAnswer(spark.sql("SELECT id FROM TNN"), Row(2) :: Nil) + } + } + + test("Fallback: partial-update with sequence-group delete keeps its semantics (scan path)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TPU (id INT, g INT, v BIGINT) + |TBLPROPERTIES ( + | 'primary-key' = 'id', + | 'bucket' = '2', + | 'merge-engine' = 'partial-update', + | 'fields.g.sequence-group' = 'v', + | 'partial-update.remove-record-on-sequence-group' = 'g') + |""".stripMargin) + spark.sql("INSERT INTO TPU VALUES (1, 1, 10), (2, 1, 20)") + + // The fast path would blank the sequence-group field g and change semantics; it must + // fall back to the scan path so the behavior stays identical to master (which currently + // keeps both rows for this configuration, see #8858). + spark.sql("DELETE FROM TPU WHERE id = 1") + checkAnswer(spark.sql("SELECT id FROM TPU ORDER BY id"), Row(1) :: Row(2) :: Nil) + } + } + + test("Subquery fast path: NULL keys in the subquery delete nothing") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TSN (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2', 'write-only' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO TSN VALUES (1,10),(2,20)") + spark.sql("CREATE TABLE SKEYS3 (id INT)") + spark.sql("INSERT INTO SKEYS3 VALUES (CAST(NULL AS INT))") + + // NULL never matches, so no -D row may be written: otherwise we would store a NULL pk + val before = totalRecordCount("TSN") + spark.sql("DELETE FROM TSN WHERE id IN (SELECT id FROM SKEYS3)") + assert(totalRecordCount("TSN") == before, "NULL keys must not produce -D rows") + + spark.sql("CALL sys.compact(table => 'test.TSN', compact_strategy => 'full')") + checkAnswer(spark.sql("SELECT id FROM TSN ORDER BY id"), Row(1) :: Row(2) :: Nil) + } + } + + test("Fallback: correlated subquery still works (scan path)") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TC (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2') + |""".stripMargin) + spark.sql("INSERT INTO TC VALUES (1,10),(2,20),(3,30)") + spark.sql("CREATE TABLE CKEYS (id INT, v BIGINT)") + spark.sql("INSERT INTO CKEYS VALUES (1,5),(2,30),(3,5)") + + // correlated: the subquery references the target table, so it can not be planned alone + spark.sql("DELETE FROM TC WHERE id IN (SELECT id FROM CKEYS c WHERE c.v > TC.v)") + checkAnswer(spark.sql("SELECT id FROM TC ORDER BY id"), Row(1) :: Row(3) :: Nil) + } + } + + test("Fallback: sequence.field falls back to scan path") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TSEQ (id INT, v BIGINT, ts BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2', + | 'sequence.field' = 'ts', 'write-only' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO TSEQ VALUES (1,10,100),(2,20,100)") + + // a NULL sequence value would sort oldest and lose the merge, so the fast path is off + assertScanPath("TSEQ", "id = 999") + + spark.sql("DELETE FROM TSEQ WHERE id = 1") + spark.sql("CALL sys.compact(table => 'test.TSEQ', compact_strategy => 'full')") + checkAnswer(spark.sql("SELECT id FROM TSEQ"), Row(2) :: Nil) + } + } + + test("Fallback: cross-partition table falls back to scan path") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + // pk does not contain the partition field -> cross partition update, the -D row would carry + // a NULL partition + spark.sql(""" + |CREATE TABLE TXP (id INT, v BIGINT, dt STRING) + |PARTITIONED BY (dt) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '-1') + |""".stripMargin) + spark.sql("INSERT INTO TXP VALUES (1,10,'p1'),(2,20,'p2')") + + spark.sql("DELETE FROM TXP WHERE id = 1") + checkAnswer(spark.sql("SELECT id, dt FROM TXP"), Row(2, "p2") :: Nil) + } + } + + test("Fallback: too many literal keys falls back to scan path") { + withSparkSQLConf( + "spark.paimon.delete.point-delete.enabled" -> "true", + "spark.paimon.delete.point-delete.max-rows" -> "1") { + spark.sql(""" + |CREATE TABLE TMAX (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2', 'write-only' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO TMAX VALUES (1,10),(2,20),(3,30)") + + assertScanPath("TMAX", "id IN (998, 999)") + + spark.sql("DELETE FROM TMAX WHERE id IN (1, 2)") + spark.sql("CALL sys.compact(table => 'test.TMAX', compact_strategy => 'full')") + checkAnswer(spark.sql("SELECT id FROM TMAX"), Row(3) :: Nil) + } + } + + test("Fast path is opt-in: disabled by default") { + spark.sql(""" + |CREATE TABLE TOFF (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2', 'write-only' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO TOFF VALUES (1,10),(2,20)") + + assertScanPath("TOFF", "id = 999") + } + + test("Point delete fast path really skips the scan") { + withSparkSQLConf("spark.paimon.delete.point-delete.enabled" -> "true") { + spark.sql(""" + |CREATE TABLE TFP (id INT, v BIGINT) + |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '2', 'write-only' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO TFP VALUES (1,10),(2,20)") + + assertFastPath("TFP", "id = 999") + + spark.sql("CALL sys.compact(table => 'test.TFP', compact_strategy => 'full')") + checkAnswer(spark.sql("SELECT id FROM TFP ORDER BY id"), Row(1) :: Row(2) :: Nil) + } + } +}