+ * The default implementation ignores {@code options} and delegates to + * {@link #deleteWhere(Predicate[])}, which preserves backward compatibility for connectors + * that do not need per-statement options. + * + * @param predicates predicate expressions, used to select rows to delete when all expressions + * match + * @param options per-statement options from the {@code WITH} clause; empty if none were given + * @throws IllegalArgumentException If the delete is rejected due to required effort + * @since 4.3.0 + */ + default void deleteWhere(Predicate[] predicates, CaseInsensitiveStringMap options) { + deleteWhere(predicates); + } + @Override default boolean truncateTable() { + return truncateTable(CaseInsensitiveStringMap.empty()); + } + + @Override + default boolean truncateTable(CaseInsensitiveStringMap options) { Predicate[] predicates = new Predicate[] { new AlwaysTrue() }; boolean canDelete = canDeleteWhere(predicates); if (canDelete) { - deleteWhere(predicates); + deleteWhere(predicates, options); } return canDelete; } diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TruncatableTable.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TruncatableTable.java index 058e105758c72..eccd9809c5e6b 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TruncatableTable.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TruncatableTable.java @@ -20,6 +20,7 @@ import org.apache.spark.annotation.Evolving; import org.apache.spark.sql.connector.metric.CustomMetric; import org.apache.spark.sql.connector.metric.CustomTaskMetric; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; /** * Represents a table which can be atomically truncated. @@ -37,6 +38,20 @@ public interface TruncatableTable extends Table { */ boolean truncateTable(); + /** + * Truncate a table with per-statement options from a {@code DELETE ... WITH (key=value)} clause. + *
+ * The default implementation ignores {@code options} and delegates to {@link #truncateTable()}, + * which preserves backward compatibility for connectors that do not need per-statement options. + * + * @param options per-statement options from the {@code WITH} clause; empty if none were given + * @return true if the table was truncated successfully, false otherwise + * @since 4.3.0 + */ + default boolean truncateTable(CaseInsensitiveStringMap options) { + return truncateTable(); + } + /** * Returns an array of supported custom metrics with name and description. * By default it returns empty array. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala index f8881e2077103..c8795b8ad9b1f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala @@ -25,7 +25,6 @@ import org.apache.spark.sql.connector.catalog.{SupportsDeleteV2, SupportsRowLeve import org.apache.spark.sql.connector.write.{RowLevelOperationTable, SupportsDelta} import org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, ExtractV2Table} -import org.apache.spark.sql.util.CaseInsensitiveStringMap /** * A rule that rewrites DELETE operations using plans that operate on individual or groups of rows. @@ -45,7 +44,7 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand { d case r @ ExtractV2Table(t: SupportsRowLevelOperations) => - val table = buildOperationTable(t, DELETE, CaseInsensitiveStringMap.empty()) + val table = buildOperationTable(t, DELETE, r.options) table.operation match { case _: SupportsDelta => buildWriteDeltaPlan(r, table, cond) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala index 032e032fa1a95..b6472bed82e6b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala @@ -1230,7 +1230,8 @@ class AstBuilder extends DataTypeAstBuilder override def visitDeleteFromTable( ctx: DeleteFromTableContext): LogicalPlan = withOrigin(ctx) { val table = createUnresolvedRelation( - ctx.identifierReference, writePrivileges = Set(TableWritePrivilege.DELETE)) + ctx.identifierReference, Option(ctx.optionsClause()), + writePrivileges = Set(TableWritePrivilege.DELETE)) val tableAlias = getTableAliasWithoutColumnAlias(ctx.tableAlias(), "DELETE") val aliasedTable = tableAlias.map(SubqueryAlias(_, table)).getOrElse(table) val predicate = if (ctx.whereClause() != null) { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala index eed6fd7ca6caf..3c66f3cf19dd5 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala @@ -35,6 +35,7 @@ import org.apache.spark.sql.connector.catalog.constraints.Constraint import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure import org.apache.spark.sql.connector.expressions.Transform import org.apache.spark.sql.connector.expressions.filter.Predicate +import org.apache.spark.sql.util.CaseInsensitiveStringMap import org.apache.spark.sql.connector.write.{DeltaWrite, RowLevelOperation, RowLevelOperationTable, SupportsDelta, Write} import org.apache.spark.sql.connector.write.RowLevelOperation.Command.{DELETE, MERGE, UPDATE} import org.apache.spark.sql.errors.DataTypeErrors.toSQLType @@ -1094,7 +1095,8 @@ case class DeleteFromTable( */ case class DeleteFromTableWithFilters( table: LogicalPlan, - condition: Seq[Predicate]) extends LeafCommand + condition: Seq[Predicate], + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()) extends LeafCommand /** * The logical plan of the UPDATE TABLE command. diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala index f8d16cb0097c1..24fbade8f3494 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala @@ -2212,6 +2212,33 @@ class DDLParserSuite extends AnalysisTest { stop = 56)) } + test("SPARK-58008: delete from table: with options") { + parseCompare( + """ + |DELETE FROM testcat.ns1.ns2.tbl WITH (`write.split-size` = 10) + |WHERE a = 1 + """.stripMargin, + DeleteFromTable( + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), + new CaseInsensitiveStringMap( + java.util.Map.of("write.split-size", "10"))), + EqualTo(UnresolvedAttribute("a"), Literal(1)))) + } + + test("SPARK-58008: delete from table: with options and alias") { + parseCompare( + """ + |DELETE FROM testcat.ns1.ns2.tbl AS t WITH (`k` = 'v') + |WHERE t.a = 2 + """.stripMargin, + DeleteFromTable( + SubqueryAlias("t", + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), + new CaseInsensitiveStringMap( + java.util.Map.of("k", "v")))), + EqualTo(UnresolvedAttribute("t.a"), Literal(2)))) + } + test("update table: basic") { parseCompare( """ diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala index f62f1b0dce25e..3fda368a5af84 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala @@ -86,7 +86,6 @@ class InMemoryRowLevelOperationTable private ( // used in row-level operation tests to verify passed records // (operation, id, metadata, row) var lastWriteLog: Seq[InternalRow] = Seq.empty - override def copy(): Table = { val copied = InMemoryRowLevelOperationTable.withColumns( name = name, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala index 9c745461b1709..bcdbf5fe6ee29 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala @@ -527,8 +527,8 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat case OverwritePartitionsDynamic(r: DataSourceV2Relation, query, _, _, _, Some(write)) => OverwritePartitionsDynamicExec(planLater(query), refreshCache(r), write, r.name) :: Nil - case DeleteFromTableWithFilters(r: DataSourceV2Relation, filters) => - DeleteFromTableExec(r.table.asDeletable, filters.toArray, refreshCache(r)) :: Nil + case DeleteFromTableWithFilters(r: DataSourceV2Relation, filters, options) => + DeleteFromTableExec(r.table.asDeletable, filters.toArray, refreshCache(r), options) :: Nil case DeleteFromTable(relation, condition) => relation match { @@ -547,11 +547,11 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat table match { case t: SupportsDeleteV2 if t.canDeleteWhere(filters) => - DeleteFromTableExec(t, filters, refreshCache(r)) :: Nil + DeleteFromTableExec(t, filters, refreshCache(r), r.options) :: Nil case t: SupportsDeleteV2 => throw QueryCompilationErrors.cannotDeleteTableWhereFiltersError(t, filters) case t: TruncatableTable if condition == TrueLiteral => - TruncateTableExec(t, refreshCache(r)) :: Nil + TruncateTableExec(t, refreshCache(r), r.options) :: Nil case _ => throw QueryCompilationErrors.tableDoesNotSupportDeletesError(table) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DeleteFromTableExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DeleteFromTableExec.scala index ea61bca266600..c13cf4e8d4023 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DeleteFromTableExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DeleteFromTableExec.scala @@ -24,11 +24,13 @@ import org.apache.spark.sql.connector.catalog.SupportsDeleteV2 import org.apache.spark.sql.connector.catalog.transactions.Transaction import org.apache.spark.sql.connector.expressions.filter.Predicate import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.util.CaseInsensitiveStringMap case class DeleteFromTableExec( table: SupportsDeleteV2, condition: Array[Predicate], refreshCache: () => Unit, + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty(), transaction: Option[Transaction] = None) extends LeafV2CommandExec with TransactionalExec @@ -42,7 +44,7 @@ case class DeleteFromTableExec( override protected def run(): Seq[InternalRow] = { try { - table.deleteWhere(condition) + table.deleteWhere(condition, options) } finally { postDriverMetrics(table.reportDriverMetrics()) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala index c02fdec1fab60..c052edea53e08 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala @@ -47,7 +47,8 @@ object OptimizeMetadataOnlyDeleteFromTable extends Rule[LogicalPlan] with Predic if (filtersOpt.exists(table.canDeleteWhere)) { logDebug(s"Switching to delete with filters: " + s"${filtersOpt.get.mkString("[", ", ", "]")}") - DeleteFromTableWithFilters(relation, filtersOpt.get.toImmutableArraySeq) + DeleteFromTableWithFilters( + relation, filtersOpt.get.toImmutableArraySeq, relation.options) } else { tryDeleteWithPartitionPredicates(table, relation, normalizedPredicates) .getOrElse { @@ -90,7 +91,7 @@ object OptimizeMetadataOnlyDeleteFromTable extends Rule[LogicalPlan] with Predic } yield { logDebug(s"Switching to delete with PartitionPredicate filters: " + s"${combined.mkString("[", ", ", "]")}") - DeleteFromTableWithFilters(relation, combined.toImmutableArraySeq) + DeleteFromTableWithFilters(relation, combined.toImmutableArraySeq, relation.options) } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TruncateTableExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TruncateTableExec.scala index 0bcfc4f364182..6c6c5f5374a9c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TruncateTableExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TruncateTableExec.scala @@ -21,13 +21,15 @@ import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.connector.catalog.TruncatableTable import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.util.CaseInsensitiveStringMap /** * Physical plan node for table truncation. */ case class TruncateTableExec( table: TruncatableTable, - refreshCache: () => Unit) + refreshCache: () => Unit, + options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()) extends LeafV2CommandExec with SupportsCustomDriverMetrics { @@ -38,7 +40,7 @@ case class TruncateTableExec( override protected def run(): Seq[InternalRow] = { try { - if (table.truncateTable()) refreshCache() + if (table.truncateTable(options)) refreshCache() Seq.empty } finally { postDriverMetrics(table.reportDriverMetrics()) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala index b894d5d75b3c8..aa289a058e1d6 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala @@ -19,12 +19,12 @@ package org.apache.spark.sql.connector import org.apache.spark.internal.config import org.apache.spark.sql.{AnalysisException, Row} +import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured import org.apache.spark.sql.catalyst.expressions.CheckInvariant -import org.apache.spark.sql.catalyst.plans.logical.Filter -import org.apache.spark.sql.connector.catalog.{Aborted, Committed} -import org.apache.spark.sql.connector.catalog.InMemoryTable -import org.apache.spark.sql.connector.write.DeleteSummary -import org.apache.spark.sql.execution.datasources.v2.{DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, ReplaceData, WriteDelta} +import org.apache.spark.sql.connector.catalog.{Aborted, Committed, InMemoryTable, RowLevelOperationWithOptions} +import org.apache.spark.sql.connector.write.{DeleteSummary, RowLevelOperationTable} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.sources @@ -1037,6 +1037,122 @@ abstract class DeleteFromTableSuiteBase extends RowLevelOperationSuiteBase { } } + protected def checkRowLevelOperationOptions( + func: => Unit, + expectedOptions: (String, String)*): Unit = { + val Seq(qe) = withQueryExecutionsCaptured(spark)(func) + val writeRelation = qe.optimizedPlan.collectFirst { + case rd: ReplaceData => rd.table + case wd: WriteDelta => wd.table + }.getOrElse(fail("couldn't find row-level operation in optimized plan")) + .asInstanceOf[DataSourceV2Relation] + val operation = writeRelation.table.asInstanceOf[RowLevelOperationTable].operation + .asInstanceOf[RowLevelOperationWithOptions] + expectedOptions.foreach { case (key, value) => + assert(writeRelation.options.get(key) === value, s"relation option '$key'") + assert(operation.options.get(key) === value, s"row-level operation option '$key'") + assert(table.lastWriteInfo.options().get(key) === value, s"write option '$key'") + } + } + + test("SPARK-58008: delete with dynamic options") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |""".stripMargin) + + // Use salary < 200 (LessThan) rather than pk = 1 (EqualTo) so the optimizer does not + // replace the row-level plan with a filter-only deleteWhere call via + // OptimizeMetadataOnlyDeleteFromTable (canDeleteWhere returns false for LessThan). + checkRowLevelOperationOptions( + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) WHERE salary < 200"), + "write.split-size" -> "10") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + + test("SPARK-58008: delete with dynamic options and a subquery on the same table") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |{ "pk": 3, "salary": 300, "dep": "hr" } + |""".stripMargin) + + checkRowLevelOperationOptions( + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) " + + s"WHERE pk IN (SELECT pk FROM $tableNameAsString WHERE dep = 'hr')"), + "write.split-size" -> "10") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + + test("SPARK-58008: delete with dynamic options and a CTE") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |{ "pk": 3, "salary": 300, "dep": "hr" } + |""".stripMargin) + + checkRowLevelOperationOptions( + sql(s"""WITH cte AS ( + | SELECT pk FROM $tableNameAsString WHERE dep = 'hr' + |) + |DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) + |WHERE pk IN (SELECT pk FROM cte)""".stripMargin), + "write.split-size" -> "10") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + + test("SPARK-58008: delete with dynamic options on the deleteWhere path") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |""".stripMargin) + + // dep = 'hr' is an EqualTo on the partition column. OptimizeMetadataOnlyDeleteFromTable + // converts the row-level plan to a deleteWhere call. Verify options flow into the exec. + val Seq(qe) = withQueryExecutionsCaptured(spark) { + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) WHERE dep = 'hr'") + } + val exec = qe.executedPlan.collectFirst { + case e: DeleteFromTableExec => e + }.getOrElse(fail("expected DeleteFromTableExec for the metadata-only deleteWhere path")) + assert(exec.options.get("write.split-size") === "10", + "options must reach DeleteFromTableExec on the deleteWhere path") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + + test("SPARK-58008: delete all rows with dynamic options") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |""".stripMargin) + + // DELETE with no WHERE: RewriteDeleteFromTable leaves the plan unrewritten (TruncatableTable + // short-circuit). DataSourceV2Strategy then matches SupportsDeleteV2.canDeleteWhere(AlwaysTrue) + // and produces DeleteFromTableExec, carrying the WITH-clause options through r.options. + val Seq(qe) = withQueryExecutionsCaptured(spark) { + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10)") + } + val exec = qe.executedPlan.collectFirst { + case e: DeleteFromTableExec => e + }.getOrElse(fail("expected DeleteFromTableExec for the delete-all path")) + assert(exec.options.get("write.split-size") === "10", + "options must reach DeleteFromTableExec on the delete-all path") + + checkAnswer(sql(s"SELECT * FROM $tableNameAsString"), Nil) + } + private def executeDeleteWithRewrite(query: String): Unit = { val executedPlan = executeAndKeepPlan { sql(query)