[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/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

@@ -1229,7 +1229,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/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/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,59 @@ 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)
+  }
+
   private def executeDeleteWithRewrite(query: String): Unit = {
     val executedPlan = executeAndKeepPlan {
       sql(query)