[SPARK-58008][SQL] Support dynamic table options for DELETE

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -744,7 +744,7 @@ resource
 dmlStatementNoWith
     : insertInto (query | LEFT_PAREN query RIGHT_PAREN queryAlias=tableAlias)      #singleInsertQuery
     | fromClause multiInsertQueryBody+                                             #multiInsertQuery
-    | DELETE FROM identifierReference tableAlias whereClause?                      #deleteFromTable
+    | DELETE FROM identifierReference tableAlias optionsClause? whereClause?       #deleteFromTable
     | UPDATE identifierReference tableAlias optionsClause? setClause whereClause?  #updateTable
     | MERGE (WITH SCHEMA EVOLUTION)? INTO target=identifierReference targetAlias=tableAlias
         USING (source=identifierReference |

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsDeleteV2.java

@@ -20,6 +20,7 @@
 import org.apache.spark.annotation.Evolving;
 import org.apache.spark.sql.connector.expressions.filter.AlwaysTrue;
 import org.apache.spark.sql.connector.expressions.filter.Predicate;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
 /**
  * A mix-in interface for {@link Table} delete support. Data sources can implement this
@@ -72,12 +73,35 @@ default boolean canDeleteWhere(Predicate[] predicates) {
    */
   void deleteWhere(Predicate[] predicates);
 
+  /**
+   * Delete data from a data source table that matches filter expressions, forwarding the
+   * per-statement options from a {@code DELETE ... WITH (key=value)} clause.
+   * <p>
+   * 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;
   }

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.
+   * <p>
+   * 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.

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)

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) {

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.

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(
       """

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,

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)
           }

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())
     }

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)
     }
   }
 

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())